挑战项目 B · 信息流模拟器：亲手养出一个信息茧房，再治好它

   你要构建：
   ① 一个模拟世界：1000 个虚拟用户，各有真实兴趣分布 + 疲劳机制
   ② 一个推荐系统：在这个世界里每天给用户推内容
   ③ 一个反馈循环：用户的点击回流成训练数据，模型每"天"重训

   然后你会亲眼看到（用自己的图表）：
   💀 第 5 天起，推荐多样性开始崩塌
   💀 第 15 天，用户看到的类目从 8 个缩到 2 个
   💀 第 20 天，模拟用户开始"流失"

   最后你要当医生：
   💊 加探索（Thompson Sampling）、加多样性（MMR/DPP）
   💊 对比治疗前后的"30 天留存曲线"

import numpy as np

N_TOPICS = 8          # 8 个内容类目：科技/体育/娱乐/美食/游戏/时政/时尚/科普
N_USERS  = 1000
N_ITEMS_PER_DAY = 200 # 每天新产生 200 条内容

class SimUser:
    def __init__(self, rng):
        # 真实兴趣：Dirichlet 采样 → 大多数人有 2-3 个主兴趣 + 若干弱兴趣
        self.interest = rng.dirichlet(alpha=np.full(N_TOPICS, 0.7))
        self.fatigue  = np.zeros(N_TOPICS)   # 各类目的疲劳度
        self.satisfaction = 0.7              # 满意度，决定会不会流失
        self.alive = True

    def click_prob(self, item_topic, item_quality):
        """点击概率 = 兴趣 × 质量 × (1 - 疲劳)"""
        base = self.interest[item_topic] * item_quality
        return np.clip(base * (1 - self.fatigue[item_topic]), 0.001, 0.95)

    def consume(self, shown_topics, clicked_mask, rng):
        """看完一屏后的状态更新——这里是模拟器的灵魂"""
        # ① 点了的类目：短期兴趣小幅上升（上瘾机制）但疲劳累积
        for t, c in zip(shown_topics, clicked_mask):
            if c:
                self.fatigue[t] = min(self.fatigue[t] + 0.08, 0.9)
            else:
                self.fatigue[t] = max(self.fatigue[t] - 0.02, 0)
        # 没被展示的类目疲劳自然恢复
        shown = set(shown_topics)
        for t in range(N_TOPICS):
            if t not in shown:
                self.fatigue[t] = max(self.fatigue[t] - 0.05, 0)

        # ② 满意度更新：既要"点得多"，也要"看得广"
        ctr_today = clicked_mask.mean()
        variety   = len(set(shown_topics)) / N_TOPICS
        self.satisfaction = 0.85 * self.satisfaction + 0.15 * (0.6*ctr_today + 0.4*variety)

        # ③ 流失判定：满意度太低就再见
        if rng.random() < max(0, 0.25 - self.satisfaction) :
            self.alive = False

class SimItem:
    def __init__(self, item_id, day, rng):
        self.id = item_id
        self.topic = rng.integers(N_TOPICS)
        self.quality = np.clip(rng.beta(2, 5) * 2, 0, 1)  # 大多数内容平庸，少数精品
        self.birth_day = day

def run_simulation(policy, days=30, seed=42):
    rng = np.random.default_rng(seed)
    users = [SimUser(rng) for _ in range(N_USERS)]
    items, logs, metrics = [], [], []

    for day in range(days):
        # 每天新内容上架（持续的物品冷启动！）
        items += [SimItem(len(items)+k, day, rng) for k in range(N_ITEMS_PER_DAY)]
        fresh = [it for it in items if day - it.birth_day <= 3]   # 只推 3 天内的

        day_stats = {"day": day, "clicks": 0, "shows": 0,
                     "topic_counts": np.zeros(N_TOPICS)}
        for uid, u in enumerate(users):
            if not u.alive:
                continue
            slate = policy.select(uid, u, fresh, k=10, day=day)   # 推 10 条
            topics = np.array([it.topic for it in slate])
            probs  = np.array([u.click_prob(it.topic, it.quality) for it in slate])
            clicks = rng.random(len(slate)) < probs

            u.consume(topics, clicks, rng)
            for it, c in zip(slate, clicks):                      # 日志回流
                logs.append((day, uid, it.id, it.topic, int(c)))
            day_stats["clicks"] += clicks.sum(); day_stats["shows"] += len(slate)
            for t in topics: day_stats["topic_counts"][t] += 1

        # 晚上：用累计日志重训模型（反馈循环的关键一环）
        policy.retrain(logs, items)

        # 记录指标
        p = day_stats["topic_counts"]; p = p / max(p.sum(), 1)
        entropy = -np.sum(p[p > 0] * np.log(p[p > 0])) / np.log(N_TOPICS)
        alive = sum(u.alive for u in users)
        metrics.append({
            "day": day, "dau": alive,
            "ctr": day_stats["clicks"] / max(day_stats["shows"], 1),
            "topic_entropy": entropy,                 # ⭐ 茧房温度计
            "satisfaction": np.mean([u.satisfaction for u in users if u.alive]),
        })
        print(metrics[-1])
    return metrics, logs

# 策略接口 —— 四种策略都实现它
class Policy:
    def select(self, uid, user, candidates, k, day): ...
    def retrain(self, logs, items): ...

class RandomPolicy(Policy):
    def __init__(self, seed=0): self.rng = np.random.default_rng(seed)
    def select(self, uid, user, cands, k, day):
        return list(self.rng.choice(cands, size=min(k, len(cands)), replace=False))
    def retrain(self, logs, items): pass