import pandas as pd, numpy as np, zipfile, urllib.request, io
from collections import defaultdict

# ---------- T1: 数据 ----------
url = "https://files.grouplens.org/datasets/movielens/ml-latest-small.zip"
with urllib.request.urlopen(url) as r:
    z = zipfile.ZipFile(io.BytesIO(r.read()))
    df = pd.read_csv(z.open("ml-latest-small/ratings.csv"))

# 只保留正反馈（评分 >= 4 视为"喜欢"）—— 转成隐式反馈问题
df = df[df.rating >= 4.0].copy()
print(f"用户 {df.userId.nunique()}, 物品 {df.movieId.nunique()}, 交互 {len(df)}")

# ---------- T2: 按时间切分（关键！） ----------
df = df.sort_values("timestamp")
split_ts = df.timestamp.quantile(0.8)
train = df[df.timestamp <= split_ts]
test  = df[df.timestamp >  split_ts]

# 测试集里只保留训练集见过的用户（冷启动单独评估）
test = test[test.userId.isin(train.userId.unique())]

train_dict = train.groupby("userId").movieId.apply(list).to_dict()
test_dict  = test.groupby("userId").movieId.apply(set).to_dict()

# ---------- T5: 评估函数（先写好，后面复用） ----------
def evaluate(recommend_fn, train_dict, test_dict, k=10):
    recalls, ndcgs, hits = [], [], []
    for user, truth in test_dict.items():
        recs = recommend_fn(user, k)
        if not recs: continue
        hit_set = set(recs) & truth
        recalls.append(len(hit_set) / len(truth))
        hits.append(1.0 if hit_set else 0.0)
        # NDCG
        dcg = sum(1/np.log2(i+2) for i, item in enumerate(recs) if item in truth)
        idcg = sum(1/np.log2(i+2) for i in range(min(len(truth), k)))
        ndcgs.append(dcg/idcg if idcg > 0 else 0)
    return {"Recall@%d" % k: np.mean(recalls),
            "NDCG@%d" % k:   np.mean(ndcgs),
            "HitRate@%d" % k: np.mean(hits)}

# ---------- T3: Baseline 1 —— 热门推荐 ----------
popular = train.movieId.value_counts().index.tolist()
def rec_popular(user, k=10):
    seen = set(train_dict.get(user, []))
    return [i for i in popular if i not in seen][:k]

print("热门推荐:", evaluate(rec_popular, train_dict, test_dict))

# ---------- T3: Baseline 2 —— ItemCF ----------
# 👉 把第 4 节的 ItemCF 类复制过来，fit(train_dict)，然后：
# print("ItemCF:", evaluate(lambda u,k: [i for i,_ in itemcf.recommend(train_dict[u], k)],
#                           train_dict, test_dict))

# ---------- 召回负采样：必须全库随机采，不是"曝光未点击" ----------
# 见第 8 节 in_batch_softmax_loss

# ---------- Faiss 索引 ----------
import faiss
item_vecs = model.encode_item(all_item_ids).detach().numpy().astype('float32')
faiss.normalize_L2(item_vecs)
index = faiss.IndexFlatIP(item_vecs.shape[1])   # 数据量小，暴力就行
index.add(item_vecs)

def retrieve(user_id, k=200):
    uv = model.encode_user(torch.tensor([user_id])).detach().numpy().astype('float32')
    faiss.normalize_L2(uv)
    _, ids = index.search(uv, k)
    return ids[0]

# ---------- 排序阶段的负样本：从召回结果里采 ----------
# ⭐ 这一点很重要：排序模型的负样本分布要接近它线上面对的分布（召回结果）
def build_rank_samples(user, pos_items, retrieved, n_neg=4):
    samples = [(user, i, 1) for i in pos_items]
    negs = [i for i in retrieved if i not in pos_items]
    samples += [(user, i, 0) for i in np.random.choice(negs, min(len(negs), len(pos_items)*n_neg), replace=False)]
    return samples

from fastapi import FastAPI
import hashlib, time, logging

app = FastAPI()
log = logging.getLogger("rec")

def assign_group(user_id: int, exp_id: str = "exp_001") -> str:
    """A/B 分流：哈希要带 exp_id，否则多个实验会用同一个分桶"""
    h = int(hashlib.md5(f"{exp_id}_{user_id}".encode()).hexdigest(), 16)
    return "treatment" if h % 100 < 50 else "control"

@app.get("/recommend")
def recommend(user_id: int, n: int = 10):
    t0 = time.time()
    group = assign_group(user_id)

    try:
        # 1. 特征
        user_feat = feature_store.get_user(user_id)

        # 2. 召回（多路）
        cands = set()
        cands |= set(vector_recall(user_feat, k=200))
        cands |= set(itemcf_recall(user_feat, k=100))
        cands |= set(hot_recall(k=50))
        cands -= set(user_feat["history"])          # 去重
        cands = list(cands)

        # 3. 排序（A/B：两组用不同模型）
        model = model_v2 if group == "treatment" else model_v1
        scored = model.predict_batch(user_id, cands)

        # 4. 重排：打散 + 截断
        result = rerank(scored, n=n)

    except Exception as e:
        log.exception("rec failed, fallback to hot")
        result = HOT_LIST[:n]                       # 🛟 兜底，绝不返回空
        group = "fallback"

    latency = (time.time() - t0) * 1000
    # 5. 埋点（生产里应该异步写 Kafka）
    log.info({"user_id": user_id, "group": group, "items": result,
              "latency_ms": round(latency, 2),
              "avg_score": float(np.mean([s for _, s in scored])) if result else None})
    return {"user_id": user_id, "items": result, "group": group,
            "latency_ms": round(latency, 2)}

@app.get("/health")
def health():
    return {"status": "ok"}

pip install implicit recbole lightfm faiss-cpu torch

# 推荐系统实践

## 问题定义
数据集、规模、目标（Top-N 推荐 / CTR 预估）、评估指标

## 数据处理
- 稀疏度、长尾分布（放图）
- 切分策略：按时间切分，避免数据泄漏

## 方法
| 模型 | Recall@10 | NDCG@10 | 训练时间 |
|------|-----------|---------|---------|
| 热门推荐 | 0.052 | 0.031 | - |
| ItemCF | 0.118 | 0.076 | 12s |
| ALS | 0.145 | 0.094 | 45s |
| 双塔+排序 | 0.178 | 0.121 | 8min |

## 关键发现
- 加入热门惩罚 α=0.5 后 Coverage 从 12% → 31%，Recall 仅降 3%
- 召回负样本从"曝光未点击"改成"全库随机"，Recall@200 提升 XX%

## 踩过的坑
1. 一开始随机切分数据 → Recall 虚高 3 倍，改成按时间切分后正常
2. ...

## 如何运行
...