本文目录导读：
我来详细介绍Python实现协同过滤推荐系统的几种方法。
基于用户的协同过滤（User-Based CF）
import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
class UserBasedCF:
def __init__(self, k=5):
self.k = k # 邻居数量
self.user_similarity = None
self.user_ratings = None
self.user_mean = None
def fit(self, ratings_matrix):
"""
ratings_matrix: DataFrame，行是用户，列是物品
"""
self.user_ratings = ratings_matrix
# 计算用户间相似度（余弦相似度）
self.user_similarity = cosine_similarity(ratings_matrix.fillna(0))
# 计算每个用户的平均评分
self.user_mean = ratings_matrix.mean(axis=1)
def predict(self, user_id, item_id):
"""预测用户对物品的评分"""
# 获取该用户的相似度向量
user_sim = self.user_similarity[user_id]
# 找到最相似的k个用户（排除自己）
similar_users = np.argsort(user_sim)[::-1][1:self.k+1]
# 获取这些相似用户对目标物品的评分
item_ratings = self.user_ratings.iloc[similar_users, item_id]
# 过滤掉没有评分的用户
valid_users = similar_users[~item_ratings.isna()]
if len(valid_users) == 0:
return self.user_mean[user_id]
# 计算加权平均评分
weights = user_sim[valid_users]
ratings = self.user_ratings.iloc[valid_users, item_id].values
# 去中心化评分：考虑用户评分偏差
# 预测评分 = 用户平均评分 + 加权评分偏差
predicted = self.user_mean[user_id] + np.sum(weights * (ratings - self.user_mean[valid_users])) / np.sum(weights)
return predicted
def recommend(self, user_id, n_items=5):
"""为用户推荐n个物品"""
# 获取用户未评分的物品
user_ratings = self.user_ratings.iloc[user_id]
unrated_items = user_ratings[user_ratings.isna()].index
# 预测评分
predictions = []
for item_id in unrated_items:
pred = self.predict(user_id, item_id)
predictions.append((item_id, pred))
# 排序并返回top-n
predictions.sort(key=lambda x: x[1], reverse=True)
return predictions[:n_items]
基于物品的协同过滤（Item-Based CF）
class ItemBasedCF:
def __init__(self, k=5):
self.k = k
self.item_similarity = None
self.ratings_matrix = None
def fit(self, ratings_matrix):
"""计算物品间相似度"""
self.ratings_matrix = ratings_matrix
# 转置矩阵，计算物品间的相似度
item_ratings = ratings_matrix.T
self.item_similarity = cosine_similarity(item_ratings.fillna(0))
def predict(self, user_id, item_id):
"""预测用户对物品的评分"""
# 获取用户已评分的物品
user_ratings = self.ratings_matrix.iloc[user_id]
rated_items = user_ratings[~user_ratings.isna()].index
if len(rated_items) == 0:
return 0
# 计算目标物品与用户已评分物品的相似度
item_sims = []
item_ratings = []
for rated_item in rated_items:
sim = self.item_similarity[item_id][rated_item]
if sim > 0:
item_sims.append(sim)
item_ratings.append(user_ratings[rated_item])
if len(item_sims) == 0:
return 0
# 前k个最相似的物品
k = min(self.k, len(item_sims))
top_k_indices = np.argsort(item_sims)[::-1][:k]
# 加权平均
weights = [item_sims[i] for i in top_k_indices]
ratings = [item_ratings[i] for i in top_k_indices]
predicted = np.sum(np.array(weights) * np.array(ratings)) / np.sum(weights)
return predicted
def recommend(self, user_id, n_items=5):
"""为用户推荐物品"""
# 获取用户未评分的物品
user_ratings = self.ratings_matrix.iloc[user_id]
unrated_items = user_ratings[user_ratings.isna()].index
# 预测评分
predictions = []
for item_id in range(len(unrated_items)):
pred = self.predict(user_id, item_id)
predictions.append((unrated_items[item_id], pred))
predictions.sort(key=lambda x: x[1], reverse=True)
return predictions[:n_items]
矩阵分解方法（SVD）
from sklearn.decomposition import TruncatedSVD
from sklearn.preprocessing import StandardScaler
class MatrixFactorizationCF:
def __init__(self, n_factors=50):
self.n_factors = n_factors
self.svd = TruncatedSVD(n_components=n_factors)
self.scaler = StandardScaler()
self.ratings_matrix = None
def fit(self, ratings_matrix):
self.ratings_matrix = ratings_matrix
# 填充缺失值为0
filled_matrix = ratings_matrix.fillna(0)
# 标准化
normalized_matrix = self.scaler.fit_transform(filled_matrix)
# SVD分解
self.user_features = self.svd.fit_transform(normalized_matrix)
self.item_features = self.svd.components_
def predict(self, user_id, item_id):
"""预测评分"""
# 重建评分
user_vec = self.user_features[user_id]
item_vec = self.item_features[:, item_id]
prediction = np.dot(user_vec, item_vec)
# 反标准化
prediction = self.scaler.inverse_transform([[prediction]])[0][0]
return prediction
def recommend(self, user_id, n_items=5):
"""推荐物品"""
predictions = []
for item_id in range(self.ratings_matrix.shape[1]):
if pd.isna(self.ratings_matrix.iloc[user_id, item_id]):
pred = self.predict(user_id, item_id)
predictions.append((item_id, pred))
predictions.sort(key=lambda x: x[1], reverse=True)
return predictions[:n_items]
完整的使用示例
# 创建示例数据
def create_sample_data():
# 用户-物品评分矩阵
data = {
'用户1': {'物品A': 5, '物品B': 3, '物品C': 4, '物品D': 4},
'用户2': {'物品A': 3, '物品B': 1, '物品C': 2, '物品D': 3},
'用户3': {'物品A': 4, '物品B': 3, '物品C': 4, '物品D': 3},
'用户4': {'物品A': 3, '物品B': 3, '物品C': 1, '物品D': 5},
'用户5': {'物品A': 1, '物品B': 5, '物品C': 5, '物品D': 2}
}
df = pd.DataFrame(data).T
df.index.name = '用户'
df.columns.name = '物品'
return df
# 使用示例
if __name__ == "__main__":
# 创建数据
ratings = create_sample_data()
print("用户-物品评分矩阵：")
print(ratings)
print("\n")
# 基于用户的协同过滤
print("=== 基于用户的协同过滤 ===")
ucf = UserBasedCF(k=3)
ucf.fit(ratings)
# 为用户0推荐
recommendations = ucf.recommend(0, n_items=3)
print(f"用户{'用户1'}的推荐结果：")
for item, score in recommendations:
print(f" 推荐物品: {item}, 预测评分: {score:.2f}")
print("\n")
# 基于物品的协同过滤
print("=== 基于物品的协同过滤 ===")
icf = ItemBasedCF(k=3)
icf.fit(ratings)
recommendations = icf.recommend(0, n_items=3)
print(f"用户{'用户1'}的推荐结果：")
for item, score in recommendations:
print(f" 推荐物品: {item}, 预测评分: {score:.2f}")
print("\n")
# 矩阵分解
print("=== 矩阵分解 ===")
mf = MatrixFactorizationCF(n_factors=3)
mf.fit(ratings)
recommendations = mf.recommend(0, n_items=3)
print(f"用户{'用户1'}的推荐结果：")
for item, score in recommendations:
print(f" 推荐物品: {item}, 预测评分: {score:.2f}")
使用Surprise库（推荐）
from surprise import Dataset, Reader, KNNBasic, SVD
from surprise.model_selection import train_test_split
from surprise import accuracy
# 使用Surprise库的高级实现
def surprise_cf_example():
# 准备数据（格式：user, item, rating）
data = {
'user': ['A', 'A', 'A', 'B', 'B', 'C', 'C', 'C', 'D', 'D'],
'item': ['i1', 'i2', 'i3', 'i2', 'i3', 'i1', 'i2', 'i4', 'i1', 'i4'],
'rating': [5, 3, 4, 2, 5, 4, 5, 3, 3, 4]
}
df = pd.DataFrame(data)
# 创建Reader
reader = Reader(rating_scale=(1, 5))
data = Dataset.load_from_df(df[['user', 'item', 'rating']], reader)
# 划分训练集和测试集
trainset, testset = train_test_split(data, test_size=0.2)
# 使用基于用户的协同过滤
algo = KNNBasic(sim_options={'user_based': True})
algo.fit(trainset)
predictions = algo.test(testset)
# 评估
print("基于用户的协同过滤:")
accuracy.rmse(predictions)
# 使用SVD
algo = SVD()
algo.fit(trainset)
predictions = algo.test(testset)
print("\nSVD矩阵分解:")
accuracy.rmse(predictions)
# 预测新用户
user_id = 'E'
item_id = 'i2'
pred = algo.predict(user_id, item_id)
print(f"\n预测用户 {user_id} 对物品 {item_id} 的评分: {pred.est:.2f}")
# 运行示例
surprise_cf_example()
关键优化技巧
- 处理冷启动问题：对新用户使用流行度推荐
- 评分归一化：减去用户平均评分消除偏差
- 相似度计算优化：使用稀疏矩阵（Scipy sparse matrices）
- 增量更新：实时更新相似度矩阵
这些方法可以根据具体场景选择使用，基于用户的CF适合用户数较少的系统，基于物品的CF更适合物品数较少的系统,而矩阵分解在大规模系统中表现更好。
