推荐系统（二）—— 协同过滤与矩阵分解

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75

import numpy as np
from scipy.stats import pearsonr


class UserBasedCF:
    """支持皮尔逊 / 余弦 / 欧氏距离的邻域协同过滤。"""

    def __init__(self, k: int = 20, similarity: str = "pearson"):
        self.k = k
        self.similarity = similarity

    def fit(self, ratings: dict[str, dict[str, float]]) -> "UserBasedCF":
        self.ratings = ratings
        self.user_mean = {u: np.mean(list(r.values())) for u, r in ratings.items()}
        self.users = list(ratings)
        self.user_idx = {u: i for i, u in enumerate(self.users)}
        n = len(self.users)
        self.sim = np.zeros((n, n))
        for i in range(n):
            for j in range(i + 1, n):
                s = self._pair_sim(self.users[i], self.users[j])
                self.sim[i, j] = self.sim[j, i] = s
        return self

    def _pair_sim(self, u: str, v: str) -> float:
        common = set(self.ratings[u]) & set(self.ratings[v])
        if len(common) < 2:
            return 0.0
        ru = np.array([self.ratings[u][i] for i in common])
        rv = np.array([self.ratings[v][i] for i in common])
        if self.similarity == "pearson":
            r, _ = pearsonr(ru, rv)
            return 0.0 if np.isnan(r) else float(r)
        if self.similarity == "cosine":
            return float(ru @ rv / (np.linalg.norm(ru) * np.linalg.norm(rv)))
        # 欧氏距离 → 落在 (0, 1] 的相似度
        return 1.0 / (1.0 + np.linalg.norm(ru - rv))

    def predict(self, user: str, item: str) -> float:
        if user not in self.ratings:
            return 3.0
        idx = self.user_idx[user]
        neighbours = [
            (v, self.sim[idx, self.user_idx[v]], self.ratings[v][item])
            for v in self.ratings
            if v != user and item in self.ratings[v] and self.sim[idx, self.user_idx[v]] > 0
        ]
        if not neighbours:
            return self.user_mean[user]
        neighbours.sort(key=lambda t: t[1], reverse=True)
        neighbours = neighbours[: self.k]
        num = sum(s * (r - self.user_mean[v]) for v, s, r in neighbours)
        den = sum(abs(s) for _, s, _ in neighbours)
        return self.user_mean[user] + num / den if den else self.user_mean[user]

    def recommend(self, user: str, n: int = 10) -> list[tuple[str, float]]:
        seen = set(self.ratings.get(user, {}))
        candidates = {i for r in self.ratings.values() for i in r} - seen
        scored = [(i, self.predict(user, i)) for i in candidates]
        scored.sort(key=lambda t: t[1], reverse=True)
        return scored[:n]


if __name__ == "__main__":
    ratings = {
        "Alice": {"肖申克": 5, "阿甘": 5, "泰坦尼克": 3, "教父": 4},
        "Bob":   {"肖申克": 5, "阿甘": 5, "当幸福": 4, "泰坦尼克": 2},
        "Carol": {"肖申克": 4, "阿甘": 4, "当幸福": 3, "泰坦尼克": 5, "教父": 5},
        "David": {"肖申克": 2, "阿甘": 1, "泰坦尼克": 4, "教父": 3},
        "Eve":   {"当幸福": 2, "泰坦尼克": 3, "教父": 4},
    }
    model = UserBasedCF(k=2).fit(ratings)
    print(f"Alice 对《当幸福》的预测分：{model.predict('Alice', '当幸福'):.2f}")
    for item, score in model.recommend("Alice", n=3):
        print(f"  {item}: {score:.2f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

import numpy as np
from collections import defaultdict


class ItemBasedCF:
    """基于物品的 CF，使用调整余弦相似度。"""

    def __init__(self, k: int = 20):
        self.k = k

    def fit(self, ratings: dict[str, dict[str, float]]) -> "ItemBasedCF":
        self.ratings = ratings
        self.user_mean = {u: np.mean(list(r.values())) for u, r in ratings.items()}
        self.item_users: dict[str, dict[str, float]] = defaultdict(dict)
        for u, items in ratings.items():
            for i, r in items.items():
                self.item_users[i][u] = r
        self.sim: dict[str, dict[str, float]] = defaultdict(dict)
        items = list(self.item_users)
        for a in range(len(items)):
            for b in range(a + 1, len(items)):
                s = self._adjusted_cosine(items[a], items[b])
                if s > 0:
                    self.sim[items[a]][items[b]] = s
                    self.sim[items[b]][items[a]] = s
        return self

    def _adjusted_cosine(self, i: str, j: str) -> float:
        common = set(self.item_users[i]) & set(self.item_users[j])
        if not common:
            return 0.0
        num = den_i = den_j = 0.0
        for u in common:
            di = self.item_users[i][u] - self.user_mean[u]
            dj = self.item_users[j][u] - self.user_mean[u]
            num += di * dj
            den_i += di * di
            den_j += dj * dj
        return num / np.sqrt(den_i * den_j) if den_i and den_j else 0.0

    def predict(self, user: str, item: str) -> float:
        if user not in self.ratings or item not in self.sim:
            return self.user_mean.get(user, 3.0)
        seen = self.ratings[user]
        rated_neighbours = [(j, s) for j, s in self.sim[item].items() if j in seen]
        if not rated_neighbours:
            return self.user_mean[user]
        rated_neighbours.sort(key=lambda t: t[1], reverse=True)
        rated_neighbours = rated_neighbours[: self.k]
        mean = self.user_mean[user]
        num = sum(s * (seen[j] - mean) for j, s in rated_neighbours)
        den = sum(abs(s) for _, s in rated_neighbours)
        return mean + num / den if den else mean

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

import numpy as np
import random


class MatrixFactorization:
    """SGD 训练的基础矩阵分解。"""

    def __init__(self, k=50, lr=0.01, reg=0.01, epochs=20, seed=0):
        self.k, self.lr, self.reg, self.epochs = k, lr, reg, epochs
        self.rng = np.random.default_rng(seed)

    def fit(self, ratings: dict[str, dict[str, float]]):
        users = list(ratings)
        items = list({i for r in ratings.values() for i in r})
        self.u_idx = {u: i for i, u in enumerate(users)}
        self.i_idx = {i: j for j, i in enumerate(items)}
        self.items = items
        self.P = self.rng.normal(0, 0.1, (len(users), self.k))
        self.Q = self.rng.normal(0, 0.1, (len(items), self.k))

        samples = [(self.u_idx[u], self.i_idx[i], r)
                   for u, row in ratings.items() for i, r in row.items()]

        for epoch in range(1, self.epochs + 1):
            random.shuffle(samples)
            sse = 0.0
            for u, i, r in samples:
                err = r - self.P[u] @ self.Q[i]
                p, q = self.P[u].copy(), self.Q[i].copy()
                self.P[u] += self.lr * (err * q - self.reg * p)
                self.Q[i] += self.lr * (err * p - self.reg * q)
                sse += err * err
            if epoch % 5 == 0:
                print(f"epoch {epoch}: RMSE={np.sqrt(sse / len(samples)):.4f}")
        return self

    def predict(self, user: str, item: str) -> float:
        if user not in self.u_idx or item not in self.i_idx:
            return 0.0
        return float(self.P[self.u_idx[user]] @ self.Q[self.i_idx[item]])

1
2

self.b_u[u] += self.lr * (err - self.reg_b * self.b_u[u])
self.b_i[i] += self.lr * (err - self.reg_b * self.b_i[i])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

class SVDpp:
    """带隐式反馈的 SVD++。"""

    def __init__(self, k=50, lr=0.005, reg=0.02, epochs=20):
        self.k, self.lr, self.reg, self.epochs = k, lr, reg, epochs

    def fit(self, ratings, implicit=None):
        # 索引用户/物品；初始化 P, Q, Y, 偏置, mu
        # 每一步：
        #   N = implicit_items[u]
        #   y_sum = Y[N].sum(0) / sqrt(len(N))
        #   pred  = mu + b_u + b_i + (P[u] + y_sum) @ Q[i]
        #   梯度回传到 P[u], Q[i], Y[N], b_u, b_i
        ...

1
2
3

P[u] += lr * (x_uij * (Q[i] - Q[j]) - reg * P[u])
Q[i] += lr * ( x_uij *  P[u]        - reg * Q[i])
Q[j] += lr * (-x_uij *  P[u]        - reg * Q[j])