推荐系统入门实战：协同过滤、矩阵分解与召回排序

全量物品（百万级）
  → 召回（粗选，几百~几千）   高速，宽松
  → 精排（CTR 预估，逐条打分） 准，模型
  → 重排（多样性/规则，几十）  业务约束
  → 展示

       物品1  物品2  物品3  物品4
用户A    5      ?      3      ?
用户B    4      2      ?      5
用户C    ?      ?      1      4

? = 未知交互（我们要预测的）
评分（显式）或点击/购买次数（隐式）

import pandas as pd

# 用户-物品-交互值 长表 → 矩阵
df = pd.DataFrame({'user': ['A','A','B','B','C','C'],
                   'item': ['i1','i3','i1','i2','i3','i4'],
                   'rating': [5,3,4,2,1,4]})
matrix = df.pivot(index='user', columns='item', values='rating')
print(matrix)

import numpy as np

def cosine(a, b):
    """余弦相似度：向量夹角，范围 [-1, 1]"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) + 1e-9)

def pearson(a, b):
    """皮尔逊：余弦 - 各自均值（去评分尺度差异）"""
    a_c, b_c = a - a.mean(), b - b.mean()
    return cosine(a_c, b_c)

def jaccard(a, b):
    """杰卡德：交集 / 并集（只看"是否交互"）"""
    return len(set(a) & set(b)) / len(set(a) | set(b))

找相似用户（与你评分向量最接近的 10 人）
→ 汇总他们喜欢的、你还没看过的物品
→ 按相似度加权排序

找到你打过高分的物品
→ 找与它们相似的物品（物品相似度预计算）
→ 推荐

import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity

# 物品相似度矩阵（对矩阵列算余弦）
item_sim = cosine_similarity(matrix.T.fillna(0))
# item_sim[i][j] = 物品 i 与 j 的相似度

def recommend_item_based(matrix, user, item_sim, top_k=5):
    """给 user 推荐：加权汇总其高评分物品的相似物品"""
    ratings = matrix.loc[user].fillna(0)
    scores = {}
    for item in ratings.index:
        if ratings[item] > 0:
            for other, sim in enumerate(item_sim[matrix.columns.get_loc(item)]):
                if matrix.loc[user, matrix.columns[other]] == 0:
                    scores[matrix.columns[other]] = \
                        scores.get(matrix.columns[other], 0) + ratings[item] * sim
    return sorted(scores, key=scores.get, reverse=True)[:top_k]

R (用户×物品) ≈ U (用户×k) × V^T (k×物品)

R[u][i] = U[u]·V[i] = 用户 u 对隐因子 k 的偏好 × 物品 i 在这些因子上的成分

# 用 Surprise 库做 SVD（最简）
from surprise import Dataset, Reader, SVD
from surprise.model_selection import train_test_split

data = Dataset.load_from_df(
    df[['user', 'item', 'rating']],
    Reader(rating_scale=(1, 5)))
trainset, testset = train_test_split(data, test_size=0.2)

model = SVD(n_factors=20, lr_all=0.005, reg_all=0.02)
model.fit(trainset)
pred = model.predict('A', 'i4')   # 预测用户 A 对物品 i4 的评分

# 物品特征：标签 / 类别 / 文本关键词
items = {'i1': ['动作', '科幻'], 'i2': ['爱情', '剧情'], 'i3': ['动作', '武侠']}

# 用户画像 = 其喜欢物品特征的加权汇总
def user_profile(user, items, ratings):
    profile = {}
    for item, r in ratings[user].items():
        for feat in items[item]:
            profile[feat] = profile.get(feat, 0) + r
    return profile

# 推荐：给物品特征与用户画像打点积

召回源（各自独立、并行）：
  热门召回      → 全局/类目热门（兜底，防冷门缺货）
  协同过滤召回  → Item-based 相似物品
  矩阵分解召回  → 用户向量 × 物品向量，Top-N 最近邻
  向量召回      → Embedding 相似（深度召回，见下）
  内容召回      → 用户画像匹配
→ 汇总、去重、截断 Top 500 送精排

import numpy as np

# 简化：user/item 因子向量（来自矩阵分解/深度模型）
user_vec = { 'A': np.array([0.9, 0.2, 0.1]) }
item_vec = { 'i1': np.array([0.8, 0.1, 0.2]), 'i4': np.array([0.7, 0.3, 0.05]) }

# 余弦相似度取 Top-N（工业用 Faiss/向量数据库加速）
def recall(user, k=10):
    sims = [(i, float(np.dot(user_vec[user], v))) for i, v in item_vec.items()]
    return sorted(sims, key=lambda x: x[1], reverse=True)[:k]

特征：用户特征（年龄/城市/历史偏好）
      物品特征（类目/价格/热度）
      交互特征（用户×物品统计、上下文）
→ 模型输出 P(点击)

# 简化：LR/GBDT 精排（特征工程见 /ml-feature-engineering/）
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import GradientBoostingClassifier

# 特征：user_age, item_cat, item_price, hist_ctr, hour...
X = feature_matrix            # (样本, 特征)
y = click_label               # 0/1

model = GradientBoostingClassifier(n_estimators=100, learning_rate=0.1)
model.fit(X_train, y_train)
rank_score = model.predict_proba(X_test)[:, 1]   # 排序分数

import numpy as np

def ndcg(relevant_ranks, k=10):
    """NDCG：排序质量——相关物品越靠前，分越高"""
    dcg = sum(rel / np.log2(i + 2) for i, rel in enumerate(relevant_ranks[:k]))
    idcg = sum(1 / np.log2(i + 2) for i in range(min(k, sum(relevant_ranks))))
    return dcg / idcg if idcg else 0

def recall_at_k(pred_items, true_items, k=10):
    """Recall@K：前 K 个推荐命中真实喜欢的比例"""
    hits = len(set(pred_items[:k]) & set(true_items))
    return hits / max(len(true_items), 1)