swing.py

1"""2Swing 召回（第10章 §10.1.4）34Swing 在 ItemCF 余弦相似度基础上增加了用户对重叠的惩罚项：5若两个用户 A、B 本身就有大量共同交互物品，则对 (i,j) 共现的贡献应打折扣，6从而避免"同一圈子"用户反复强化热门物品对之间的虚假相似度。78Swing 相似度：9    S(i, j) = Σ_{u,v ∈ N(i)∩N(j)} 1 / (α + |N(u) ∩ N(v)|)10"""1112from collections import defaultdict131415def build_swing_similarity(16    user_items: dict[str, list[str]],17    alpha: float = 5.0,18    top_k: int = 20,19) -> dict[str, list[tuple[str, float]]]:20    """21    离线计算 Swing 物品相似度。2223    Args:24        user_items: {user_id: [item_id, ...]}25        alpha: 平滑系数，防止分母为零26        top_k: 每个物品保留的 Top-K 邻居27    """28    # 物品→用户集合29    item_users: dict[str, set[str]] = defaultdict(set)30    # 用户→物品集合（方便计算用户间交集）31    user_item_set: dict[str, set[str]] = {u: set(items) for u, items in user_items.items()}3233    for user, items in user_items.items():34        for item in items:35            item_users[item].add(user)3637    item_sim: dict[str, dict[str, float]] = defaultdict(lambda: defaultdict(float))38    items = list(item_users.keys())3940    for idx, i in enumerate(items):41        for j in items[idx + 1 :]:42            common_users = item_users[i] & item_users[j]43            if not common_users:44                continue45            score = 0.046            common_list = list(common_users)47            for a in range(len(common_list)):48                for b in range(a + 1, len(common_list)):49                    u, v = common_list[a], common_list[b]50                    uv_common = len(user_item_set[u] & user_item_set[v])51                    score += 1.0 / (alpha + uv_common)52            item_sim[i][j] += score53            item_sim[j][i] += score5455    result: dict[str, list[tuple[str, float]]] = {}56    for item, neighbors in item_sim.items():57        ranked = sorted(neighbors.items(), key=lambda x: x[1], reverse=True)58        result[item] = ranked[:top_k]59    return result606162def recall_swing(63    user_id: str,64    user_items: dict[str, list[str]],65    item_sim: dict[str, list[tuple[str, float]]],66    top_n: int = 10,67) -> list[tuple[str, float]]:68    """在线召回：与 ItemCF 召回逻辑相同，相似度表换成 Swing 版本。"""69    history = set(user_items.get(user_id, []))70    score: dict[str, float] = defaultdict(float)71    for j in history:72        for i, sim in item_sim.get(j, []):73            if i not in history:74                score[i] += sim75    return sorted(score.items(), key=lambda x: x[1], reverse=True)[:top_n]767778# ── 快速验证 ──────────────────────────────────────────────────────────────────79if __name__ == "__main__":80    user_items = {81        "u1": ["i1", "i2", "i3"],82        "u2": ["i1", "i2", "i4"],83        "u3": ["i2", "i3", "i5"],84        "u4": ["i1", "i3", "i5"],85        "u5": ["i4", "i5"],86    }87    sim = build_swing_similarity(user_items, alpha=5.0, top_k=5)88    print("Swing 相似度（Top-3）：")89    for item in ["i1", "i2"]:90        print(f"  {item}: {sim.get(item, [])[:3]}")9192    result = recall_swing("u1", user_items, sim, top_n=5)93    print("\nu1 的 Swing 召回结果：")94    for item, s in result:95        print(f"  {item}: {s:.4f}")