电商搜索推荐系统案例：从关键词到转化闭环

# 定时全量同步，延迟高，大促期间完全不可用
@celery.task
def sync_all_products_to_es():
    products = db.query(Product).filter(Product.is_active == True).all()
    for p in products:  # 200万条逐条写入
        es.index(index='products', id=p.id, body=product_to_doc(p))

# 关键事件驱动：价格变更、库存变更走独立高优先级通道
# 通过 binlog 监听 + 消息队列分级，保证关键数据 3 秒内到达索引
def on_price_change(event: PriceChangeEvent):
    # 高优先级队列，独立 consumer 组，不参与全量同步排队
    high_priority_queue.publish({
        'type': 'price_update',
        'sku_id': event.sku_id,
        'new_price': event.new_price,
        'promotion_id': event.promotion_id,
        'timestamp': event.timestamp,
    })
 
def on_inventory_change(event: InventoryChangeEvent):
    # 库存归零的商品立即从搜索索引标记为不可售
    if event.stock == 0:
        es.update(
            index='products',
            id=event.sku_id,
            body={'doc': {'is_purchasable': False, 'updated_at': event.timestamp}}
        )

# 简单去重，粗暴去掉同品牌商品，导致相关结果丢失
def rerank(results):
    seen_brands = set()
    final = []
    for item in results:
        if item.brand not in seen_brands:
            final.append(item)
            seen_brands.add(item.brand)
    return final  # 前 20 条可能只剩 5-6 个结果

# MMR (Maximal Marginal Relevance) 平衡相关性和多样性
# lambda 控制两者权重，0.7 表示偏重相关性但保留多样性
def mmr_rerank(candidates, query_vec, item_vecs, lambda_param=0.7, top_k=20):
    selected = []
    remaining = list(range(len(candidates)))
 
    while len(selected) < top_k and remaining:
        best_score = -float('inf')
        best_idx = None
        for idx in remaining:
            # 相关性得分：与 query 的余弦相似度
            relevance = cosine_similarity(query_vec, item_vecs[idx])
            # 多样性惩罚：与已选商品的最高相似度
            max_sim = max(
                cosine_similarity(item_vecs[idx], item_vecs[s])
                for s in selected
            ) if selected else 0
            # MMR 综合得分
            score = lambda_param * relevance - (1 - lambda_param) * max_sim
            if score > best_score:
                best_score = score
                best_idx = idx
        selected.append(best_idx)
        remaining.remove(best_idx)
    return [candidates[i] for i in selected]

# 坏做法：实验分流和训练管道各自独立，信号互相污染
def assign_experiment(user_id):
    bucket = hash(user_id) % 100
    return 'treatment' if bucket < 50 else 'control'
 
# 训练管道不感知实验分组，所有用户行为写入同一份特征表
def log_click(user_id, item_id):
    feature_store.update(user_id, click_history=[item_id])  # 全局写入，不区分实验组

# 好做法：实验 ID 贯穿分流、埋点、特征、训练全链路
def assign_experiment(user_id, experiment_id='rank_v2'):
    bucket = hash(f"{user_id}:{experiment_id}") % 100
    return 'treatment' if bucket < 50 else 'control'
 
def log_click(user_id, item_id, experiment_id, group):
    # 带实验标签的行为日志，训练时按组隔离
    feature_store.update(
        user_id,
        click_history=[{
            'item_id': item_id,
            'experiment_id': experiment_id,
            'group': group,  # treatment 或 control
            'timestamp': time.time(),
        }],
    )
 
def build_training_data(experiment_id):
    # 训练数据只取该实验组内用户的行为，避免信号交叉
    return feature_store.query(
        filter={'experiment_id': experiment_id, 'group': 'treatment'}
    )

# 实时层：库存状态（秒级）—— 直接硬过滤，不走模型
def filter_by_availability(candidates):
    """库存归零的商品必须在排序前就被过滤掉，
    否则模型给了高分也没用，用户点进去买不了"""
    live_stock = stock_cache.batch_get([c.sku_id for c in candidates])
    return [c for c in candidates if live_stock[c.sku_id] > 0]
 
# 准实时层：价格/促销（分钟级）—— 作为排序特征注入
def inject_price_features(candidates, user_context):
    """价格不用于硬过滤，而是作为排序信号。
    用户对价格的敏感程度不同，模型自己学习权重"""
    price_info = price_cache.batch_get([c.sku_id for c in candidates])
    for c in candidates:
        c.features['price'] = price_info[c.sku_id]['current_price']
        c.features['has_promotion'] = price_info[c.sku_id]['is_promoted']
        c.features['discount_rate'] = price_info[c.sku_id].get('discount_rate', 0)
    return candidates

# 新品探索：在重排阶段强制插入一定比例的新品
# exploration_ratio 通常为 5%-10%
def inject_new_items(ranked_results, new_items, exploration_ratio=0.08):
    total_slots = len(ranked_results)
    new_slots = max(1, int(total_slots * exploration_ratio))
    # 新品按类目均匀分布，避免全是同一品类
    sampled = stratified_sample(new_items, n=new_slots, strata='category')
    # 插入到第 4、8、12... 位，避免头部位置被新品占据
    positions = [i for i in range(3, total_slots, 4)][:len(sampled)]
    for item, pos in zip(sampled, positions):
        ranked_results.insert(pos, item)
    return ranked_results[:total_slots]