推荐系统实战：从协同过滤到深度学习推荐

import numpy as np
from collections import defaultdict

class ItemBasedCF:
    """基于物品的协同过滤推荐系统"""
    
    def __init__(self, k=20):
        self.k = k  # 取最相似的 k 个物品
        self.item_sim = {}  # 物品相似度矩阵
        self.user_items = defaultdict(set)  # 用户交互过的物品
        self.item_users = defaultdict(set)  # 物品被哪些用户交互过
    
    def fit(self, interactions):
        """训练：计算物品共现矩阵和相似度
        interactions: [(user_id, item_id, rating), ...]
        """
        # 构建用户-物品映射
        for user, item, rating in interactions:
            self.user_items[user].add(item)
            self.item_users[item].add(user)
        
        # 计算物品共现矩阵（同时出现的用户数）
        item_cooccurrence = defaultdict(lambda: defaultdict(int))
        for item, users in self.item_users.items():
            for u in users:
                for other_item in self.user_items[u]:
                    if item != other_item:
                        item_cooccurrence[item][other_item] += 1
        
        # 计算余弦相似度
        for item_i, cooccur in item_cooccurrence.items():
            n_i = len(self.item_users[item_i])
            for item_j, count in cooccur.items():
                n_j = len(self.item_users[item_j])
                # 余弦相似度 = 共现数 / sqrt(|Ni| * |Nj|)
                self.item_sim[(item_i, item_j)] = count / np.sqrt(n_i * n_j)
    
    def recommend(self, user_id, top_n=10):
        """为用户推荐物品"""
        scores = defaultdict(float)
        for item_i in self.user_items[user_id]:
            for item_j, sim in self.item_sim.items():
                if item_j[0] == item_i and item_j[1] not in self.user_items[user_id]:
                    scores[item_j[1]] += sim
        # 按得分排序，返回 top_n
        return sorted(scores.items(), key=lambda x: x[1], reverse=True)[:top_n]

import numpy as np

class MatrixFactorization:
    """矩阵分解推荐模型 - SGD 优化"""
    
    def __init__(self, n_users, n_items, n_factors=64, lr=0.01, reg=0.02):
        self.n_factors = n_factors  # 隐向量维度
        self.lr = lr  # 学习率
        self.reg = reg  # L2 正则化系数
        
        # 随机初始化用户和物品隐向量
        np.random.seed(42)
        self.P = np.random.normal(0, 0.1, (n_users, n_factors))
        self.Q = np.random.normal(0, 0.1, (n_items, n_factors))
        self.b_u = np.zeros(n_users)  # 用户偏置
        self.b_i = np.zeros(n_items)  # 物品偏置
        self.mu = 0  # 全局平均评分
    
    def fit(self, interactions, epochs=20):
        """训练：interactions = [(user, item, rating), ...]"""
        self.mu = np.mean([r for _, _, r in interactions])
        
        for epoch in range(epochs):
            np.random.shuffle(interactions)
            total_loss = 0
            
            for user, item, rating in interactions:
                # 预测评分
                pred = self.mu + self.b_u[user] + self.b_i[item] + np.dot(self.P[user], self.Q[item])
                
                # 计算误差
                e = rating - pred
                total_loss += e ** 2
                
                # SGD 更新
                self.b_u[user] += self.lr * (e - self.reg * self.b_u[user])
                self.b_i[item] += self.lr * (e - self.reg * self.b_i[item])
                self.P[user] += self.lr * (e * self.Q[item] - self.reg * self.P[user])
                self.Q[item] += self.lr * (e * self.P[user] - self.reg * self.Q[item])
            
            rmse = np.sqrt(total_loss / len(interactions))
            print(f"Epoch {epoch+1}: RMSE = {rmse:.4f}")
    
    def predict(self, user, item):
        """预测用户对物品的评分"""
        return self.mu + self.b_u[user] + self.b_i[item] + np.dot(self.P[user], self.Q[item])

import torch
import torch.nn as nn
import torch.nn.functional as F

class TwoTowerModel(nn.Module):
    """双塔模型 - 向量召回核心"""
    
    def __init__(self, user_feature_dim, item_feature_dim, 
                 embedding_dim=128, hidden_dim=256):
        super().__init__()
        
        # 用户塔
        self.user_tower = nn.Sequential(
            nn.Linear(user_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, embedding_dim)
        )
        
        # 物品塔
        self.item_tower = nn.Sequential(
            nn.Linear(item_feature_dim, hidden_dim),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(hidden_dim, hidden_dim // 2),
            nn.ReLU(),
            nn.Linear(hidden_dim // 2, embedding_dim)
        )
    
    def forward(self, user_features, item_features):
        """计算用户-物品相似度"""
        user_vec = F.normalize(self.user_tower(user_features), dim=1)
        item_vec = F.normalize(self.item_tower(item_features), dim=1)
        # 余弦相似度（已归一化，点积即余弦相似度）
        similarity = torch.sum(user_vec * item_vec, dim=1)
        return similarity
    
    def get_user_embedding(self, user_features):
        """离线预计算物品向量后，在线只需调用此方法"""
        return F.normalize(self.user_tower(user_features), dim=1)
    
    def get_item_embedding(self, item_features):
        """离线预计算所有物品向量"""
        return F.normalize(self.item_tower(item_features), dim=1)
    
    def info_nce_loss(self, user_features, pos_item_features, 
                      neg_item_features, temperature=0.07):
        """InfoNCE 损失：拉近正样本，推远负样本"""
        user_vec = self.get_user_embedding(user_features)
        pos_vec = self.get_item_embedding(pos_item_features)
        neg_vec = self.get_item_embedding(neg_item_features)
        
        pos_sim = torch.sum(user_vec * pos_vec, dim=1) / temperature
        neg_sim = torch.sum(user_vec.unsqueeze(1) * neg_vec, dim=2) / temperature
        
        logits = torch.cat([pos_sim.unsqueeze(1), neg_sim], dim=1)
        labels = torch.zeros(user_features.size(0), dtype=torch.long)
        return F.cross_entropy(logits, labels)

import torch
import torch.nn as nn

class DeepFM(nn.Module):
    """DeepFM 推荐模型 - 精排层核心"""
    
    def __init__(self, field_dims, embed_dim=16, mlp_dims=(256, 128, 64)):
        super().__init__()
        self.num_fields = len(field_dims)
        
        # Embedding 层（FM 和 DNN 共享）
        self.embedding = nn.Embedding(sum(field_dims), embed_dim)
        # 一阶偏置
        self.fc = nn.Embedding(sum(field_dims), 1)
        
        # FM 部分：二阶交互
        self.fm_first = nn.ModuleList([
            nn.Embedding(dim, embed_dim) for dim in field_dims
        ])
        
        # DNN 部分：高阶交互
        input_dim = self.num_fields * embed_dim
        mlp_layers = []
        for dim in mlp_dims:
            mlp_layers.extend([
                nn.Linear(input_dim, dim),
                nn.BatchNorm1d(dim),
                nn.ReLU()
            ])
            input_dim = dim
        mlp_layers.append(nn.Linear(mlp_dims[-1], 1))
        self.mlp = nn.Sequential(*mlp_layers)
        self.dropout = nn.Dropout(0.3)
    
    def forward(self, x):
        """x: [batch_size, num_fields] 稀疏特征索引"""
        # 一阶部分
        first_order = self.fc(x).sum(dim=1)
        
        # FM 二阶部分
        feature_embeds = []
        offset = 0
        for dim, emb_layer in zip([1]*self.num_fields, self.fm_first):
            feature_embeds.append(self.embedding(x[:, offset:offset+1]).squeeze(1))
            offset += 1
        feature_embeds = torch.stack(feature_embeds, dim=1)
        
        # FM 交叉项
        sum_squared = torch.sum(feature_embeds, dim=1) ** 2
        squared_sum = torch.sum(feature_embeds ** 2, dim=1)
        fm_second = 0.5 * torch.sum(sum_squared - squared_sum, dim=1, keepdim=True)
        
        # DNN 部分
        dnn_input = feature_embeds.view(-1, self.num_fields * feature_embeds.size(-1))
        dnn_output = self.mlp(self.dropout(dnn_input))
        
        # 最终预测
        return first_order + fm_second + dnn_output

import numpy as np

class ThompsonSampling:
    """Thompson Sampling 多臂老虎机 - 冷启动探索策略"""
    
    def __init__(self, n_items):
        self.n_items = n_items
        # 每个物品的 Beta 分布参数
        # alpha = 成功次数 + 1, beta = 失败次数 + 1
        self.alpha = np.ones(n_items)
        self.beta = np.ones(n_items)
    
    def select_item(self):
        """从每个物品的 Beta 分布中采样，选择最大值对应的物品"""
        samples = np.random.beta(self.alpha, self.beta)
        return np.argmax(samples)
    
    def update(self, item_id, reward):
        """更新物品参数：reward=1 表示用户交互，0 表示未交互"""
        if reward == 1:
            self.alpha[item_id] += 1
        else:
            self.beta[item_id] += 1
    
    def get_exploration_rate(self, item_id):
        """获取物品当前的探索程度（方差越大说明探索空间越大）"""
        mean = self.alpha[item_id] / (self.alpha[item_id] + self.beta[item_id])
        var = (self.alpha[item_id] * self.beta[item_id]) / ((self.alpha[item_id] + self.beta[item_id]) ** 2 * (self.alpha[item_id] + self.beta[item_id] + 1))
        return mean, var

# 冷启动混合推荐策略
def cold_start_recommend(user_profile, new_items, alpha=0.7):
    """新物品的混合推荐，alpha 为内容推荐权重"""
    scores = []
    for item in new_items:
        content_score = compute_content_similarity(user_profile, item)
        cf_score = compute_cf_score(user_profile, item)
        final_score = alpha * content_score + (1 - alpha) * cf_score
        scores.append((item, final_score))
    return sorted(scores, key=lambda x: x[1], reverse=True)