Python案例如何实现推荐协同过滤

wen python案例 2026-07-17 200

本文目录导读：

Python案例如何实现推荐协同过滤

基于用户的协同过滤（User-Based CF）
基于物品的协同过滤（Item-Based CF）
矩阵分解方法（SVD）
完整的使用示例
使用Surprise库（推荐）
关键优化技巧

我来详细介绍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()