求职竞争力提升：后端、数据与算法方向的Python实战项目选题

技术求职与实战指南 阅读时长：约 12 分钟 技术栈：FastAPI / Celery / Polars / DuckDB / RAG / Docker

投递数十份简历后，很多 Python 求职者面临相同的困境：简历上明明写满了“在线图书商城”、“某知名电影爬虫”以及“鸢尾花/房价预测模型”，却在初筛阶段频频碰壁；即便获得面试机会，当面试官追问“当系统遭遇秒杀高并发流量时数据库如何防超卖”、“海量增量数据采集时如何应对分布式反爬与网络波动”、“模型在垂直领域的回答出现幻觉时如何系统性归因”时，应答往往因缺乏工程深水区经验而陷入停滞。本文旨在打破同质化 Demo 困局，直接拆解后端、数据与算法三大技术方向的高含金量实战选题，提供具体的系统架构、关键技术落地代码与 STAR 法则量化表达方案。

下面是秒杀核心路由的完整实现。通过 Lua 脚本将“检查库存”与“扣减库存”打包在 Redis 单线程中原子执行，避免了高并发竞态条件：

import asyncio
import redis.asyncio as aioredis
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel, Field

app = FastAPI(title="High Concurrency Flash Sale Engine")

# 初始化 Redis 异步连接池（实际生产需配置合理的 max_connections）
redis_pool = aioredis.ConnectionPool.from_url(
    "redis://localhost:6379/0", decode_responses=True, max_connections=50
)
redis_client = aioredis.Redis(connection_pool=redis_pool)

# Lua 原子扣减库存脚本
# KEYS[1]: 库存键名 (如 stock:goods_1001)
# ARGV[1]: 扣减数量 (如 1)
# 返回值: 1 扣减成功; 0 库存不足; -1 商品不存在
SEC_KILL_LUA_SCRIPT = """
local stock_key = KEYS[1]
local buy_num = tonumber(ARGV[1])

local current_stock = redis.call('get', stock_key)
if not current_stock then
    return -1
end

if tonumber(current_stock) >= buy_num then
    redis.call('decrby', stock_key, buy_num)
    return 1
else
    return 0
end
"""

class OrderRequest(BaseModel):
    user_id: int = Field(..., description="用户ID")
    goods_id: int = Field(..., description="商品ID")
    quantity: int = Field(1, ge=1, le=5, description="购买数量")

@app.post("/api/v1/seckill/order", status_code=status.HTTP_200_OK)
async def process_seckill_order(order: OrderRequest):
    stock_key = f"stock:goods_{order.goods_id}"
    
    # 执行 Lua 脚本保证原子扣减
    result = await redis_client.eval(SEC_KILL_LUA_SCRIPT, 1, stock_key, order.quantity)
    
    if result == -1:
        raise HTTPException(status_code=404, detail="秒杀商品不存在或活动未初始化")
    elif result == 0:
        raise HTTPException(status_code=400, detail="商品库存已售罄")
    elif result == 1:
        # 扣减成功，将订单创建任务异步推入消息队列削峰（以模拟后台异步任务为例）
        asyncio.create_task(dispatch_order_to_mq(order.user_id, order.goods_id, order.quantity))
        return {"status": "success", "message": "排队成功，订单正在生成中"}
    else:
        raise HTTPException(status_code=500, detail="系统并发异常")

async def dispatch_order_to_mq(user_id: int, goods_id: int, quantity: int):
    # 模拟推入 RabbitMQ / Celery 任务队列
    await asyncio.sleep(0.01)
    # 此处完成落库并记录流水

下面是数仓从 ODS 到 DWD/ADS 的高性能向量化清洗代码示例，展示如何用 Polars 处理缺失值、解析嵌套数据并计算窗口指标：

import polars as pl
import duckdb

def transform_ods_to_ads(raw_json_records: list[dict]) -> pl.DataFrame:
    # 1. 加载并定义严格的 Schema，利用 Polars 快速解析
    df_raw = pl.DataFrame(raw_json_records)

    # 2. DWD 清洗转换：剔除无效价格、解析时间戳、文本去噪
    df_dwd = (
        df_raw
        .filter(pl.col("raw_price").is_not_null() & (pl.col("raw_price") > 0))
        .with_columns([
            pl.col("item_id").cast(pl.Utf8),
            pl.col("raw_price").cast(pl.Float64).alias("price"),
            pl.col("timestamp").str.to_datetime("%Y-%m-%d %H:%M:%S").alias("crawl_time"),
            pl.col("title").str.strip_chars().str.to_lowercase().alias("clean_title")
        ])
    )

    # 3. 计算分析指标：计算同款商品近 30 天滑动窗口最低价与价格波动率
    df_ads = (
        df_dwd
        .sort(["item_id", "crawl_time"])
        .with_columns([
            pl.col("price").min().over("item_id").alias("history_lowest_price"),
            pl.col("price").rolling_mean(window_size=7).over("item_id").alias("rolling_7d_avg")
        ])
        .with_columns(
            ((pl.col("price") - pl.col("history_lowest_price")) / pl.col("history_lowest_price"))
            .round(4)
            .alias("price_markup_ratio")
        )
    )

    # 4. 无缝对接到 DuckDB 进行高速 SQL 聚合分析与持久化
    con = duckdb.connect(database=":memory:")
    con.register("ads_goods_prices", df_ads.to_arrow())
    
    # 提取跌破 30 天均价 15% 以上的商品报警清单
    alert_summary = con.execute("""
        SELECT item_id, clean_title, price, rolling_7d_avg, price_markup_ratio
        FROM ads_goods_prices
        WHERE price < (rolling_7d_avg * 0.85)
        ORDER BY price_markup_ratio ASC
        LIMIT 50
    """).pl()
    
    return alert_summary

通过原生 Python 协同向量与倒排索引实现混合召回与排名融合，不依赖封闭黑盒框架：

from typing import List, Dict
import math

class HybridSearchEngine:
    def __init__(self, k_constant: int = 60):
        # k_constant 用于平滑 RRF 排名倒数权重
        self.k = k_constant

    def reciprocal_rank_fusion(
        self, 
        dense_results: List[Dict], 
        sparse_results: List[Dict], 
        top_n: int = 5
    ) -> List[Dict]:
        """
        计算双路检索结果的 RRF 综合得分:
        RRF_Score(d) = sum(1 / (k + rank_i(d)))
        """
        rrf_scores: Dict[str, float] = {}
        doc_store: Dict[str, Dict] = {}

        # 处理稠密向量检索排名
        for rank, item in enumerate(dense_results, start=1):
            doc_id = item["doc_id"]
            doc_store[doc_id] = item
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (self.k + rank))

        # 处理 BM25 稀疏检索排名
        for rank, item in enumerate(sparse_results, start=1):
            doc_id = item["doc_id"]
            if doc_id not in doc_store:
                doc_store[doc_id] = item
            rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + (1.0 / (self.k + rank))

        # 根据最终融合分数降序排序
        sorted_docs = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
        
        final_top = []
        for doc_id, score in sorted_docs[:top_n]:
            entry = doc_store[doc_id].copy()
            entry["rrf_score"] = round(score, 5)
            final_top.append(entry)
            
        return final_top