#字节推荐架构

# 基于用户的协同过滤（简化版）
def recommend_for_user(user_id, n=20):
    # 1. 找到与该用户相似的用户
    similar_users = find_similar_users(user_id, top_k=100)

    # 2. 获取相似用户喜欢的文章
    candidate_articles = []
    for u in similar_users:
        articles = get_user_liked_articles(u)
        for article in articles:
            # 加权：相似度 × 用户对文章的评分
            score = user_similarity[user_id][u] * article_score[u][article]
            candidate_articles.append((article, score))

    # 3. 按分数排序，取 Top N
    candidate_articles.sort(key=lambda x: x[1], reverse=True)
    return candidate_articles[:n]

# 用户特征
user_features = {
    "user_id": "123456",
    "age": 28,
    "gender": "male",
    "city": "beijing",
    "device": "iphone",
    "interest_categories": ["科技", "体育", "财经"],  # 从历史行为推断
    "reading_history": ["article_1", "article_2", ...],  # 最近 N 天看过
    "active_hours": [9, 10, 11, 14, 15, 20, 21],  # 活跃时段
}

# 文章特征
article_features = {
    "article_id": "789",
    "author_id": "author_456",
    "category": "科技",
    "tags": ["AI", "大模型", "ChatGPT"],
    "publish_time": "2024-03-15 10:00",
    "word_count": 2000,
    "has_video": False,
    "has_images": True,
    "title_length": 25,
}

# Lambda 架构：双流融合

class LambdaRecommender:

    def recommend(self, user_id, context):
        # 1. 离线候选：从天级全量候选集中召回
        offline_candidates = self.offline_index.get(
            user_id, top_k=1000)

        # 2. 实时召回：从实时行为队列中召回
        # 用户最近 1 小时的行为更「热」
        realtime_candidates = self.realtime_index.get(
            user_id, top_k=100, recency_weight=0.3)

        # 3. 合并候选集
        all_candidates = merge_and_dedup(
            offline_candidates, realtime_candidates)

        # 4. 精排：模型打分
        scores = self.ranker.predict(user_id, all_candidates)

        # 5. 取 Top N
        return sorted(scores, key=lambda x: x.score, reverse=True)[:20]

# 实时索引：基于 Kafka 事件实时更新
class RealtimeIndexUpdater:

    def on_user_action(self, event):
        """
        event: {
            "user_id": "123",
            "article_id": "789",
            "action": "read",
            "duration": 120,  # 阅读时长（秒）
            "timestamp": 1710000000
        }
        """
        # 1. 更新用户实时特征
        self.update_user_features(event)

        # 2. 更新文章的实时统计（播放量、互动率）
        self.update_article_stats(event)

        # 3. 实时召回索引增量更新
        if event['action'] == 'read' and event['duration'] > 30:
            # 阅读超过 30 秒，视为正向行为
            self.realtime_index.add_positive(
                event['user_id'], event['article_id'])

# A/B 测试框架
class ABTest:

    def get_experiment_group(self, user_id, experiment_name):
        """
        哈希用户 ID，确保同一用户始终进入同一实验组
        """
        hash_key = f"{user_id}:{experiment_name}:salt_2024"
        hash_value = hashlib.md5(hash_key.encode()).hexdigest()
        bucket = int(hash_value, 16) % 1000  # 0-999

        experiment = self.experiment_config[experiment_name]
        # 10% 用户进入实验组
        if bucket < experiment['treatment_ratio'] * 1000:
            return 'treatment'
        else:
            return 'control'

    def run_recommend_experiment(self, user_id):
        group = self.get_experiment_group(user_id, "recommend_v3")

        if group == 'treatment':
            # 新算法：使用 DeepFM 模型
            return self.deepfm_ranker.recommend(user_id)
        else:
            # 旧算法：使用 Logistic Regression
            return self.lr_ranker.recommend(user_id)

# DeepFM 模型（PyTorch 实现）
import torch
import torch.nn as nn

class DeepFM(nn.Module):
    def __init__(self, feature_dims, embed_dim=16, hidden_dims=[256, 128, 64]):
        super().__init__()

        # 1. 特征嵌入层
        self.embeddings = nn.ModuleList([
            nn.Embedding(dim, embed_dim) for dim in feature_dims
        ])

        # 2. FM 部分：一阶特征 + 二阶交叉
        self.first_order = nn.ModuleList([
            nn.Embedding(dim, 1) for dim in feature_dims
        ])

        # 3. Deep 部分：高阶特征交叉
        layers = []
        input_dim = len(feature_dims) * embed_dim
        for hidden_dim in hidden_dims:
            layers.append(nn.Linear(input_dim, hidden_dim))
            layers.append(nn.ReLU())
            layers.append(nn.Dropout(0.2))
            input_dim = hidden_dim
        self.deep = nn.Sequential(*layers)

        # 4. 输出层
        self.output_layer = nn.Linear(hidden_dims[-1] + len(feature_dims) + 1, 1)

    def forward(self, feature_ids, feature_values):
        # feature_ids: [batch, num_features] - 特征 ID
        # feature_values: [batch, num_features] - 特征值

        # FM 一阶项
        first_orders = [self.first_order[i](feature_ids[:, i])
                        for i in range(len(self.embeddings))]
        first_order_sum = sum(first_orders)  # [batch, 1]

        # FM 二阶项：embedding 的点积和
        embeddings = [self.embeddings[i](feature_ids[:, i])
                       for i in range(len(self.embeddings))]
        # sum-of-square - square-of-sum，优化计算
        sum_square = sum(embeddings) ** 2
        square_sum = sum([e ** 2 for e in embeddings])
        second_order = (sum_square - square_sum).sum(dim=-1, keepdim=True) * 0.5

        # Deep 部分
        deep_input = torch.cat(embeddings, dim=-1)
        deep_output = self.deep(deep_input)

        # 合并 + 输出
        concat = torch.cat([first_order_sum, second_order, deep_output], dim=-1)
        output = torch.sigmoid(self.output_layer(concat))

        return output

# 在线学习：FTRL 实时更新模型参数
class OnlineLearning:

    def __init__(self, learning_rate=0.01):
        self.lr = learning_rate
        self.model = {}  # feature_id -> weight

    def update(self, features, label, pred):
        """
        features: Dict[feature_id, feature_value]
        label: 1 (click) or 0 (no click)
        pred: 模型预测值
        """
        error = label - pred  # 预测误差

        for feature_id, feature_value in features.items():
            # FTRL 更新公式
            if feature_id not in self.model:
                self.model[feature_id] = 0.0

            grad = error * feature_value  # 梯度
            self.model[feature_id] += self.lr * grad

            # 推送到在线模型服务
            model_service.update_feature(feature_id, self.model[feature_id])

    def predict(self, features):
        return sum(
            self.model.get(fid, 0.0) * fval
            for fid, fval in features.items()
        )