Python脚本数据推荐系统协同过滤如何实现

wen 实用脚本 2026-07-18 2

本文目录导读：

基于用户的协同过滤
基于物品的协同过滤
矩阵分解方法（SVD）
使用Surprise库的简单实现
使用示例
优化建议

我来详细解释Python中实现协同过滤推荐系统的几种主要方法：

基于用户的协同过滤

核心思想

找到与目标用户相似的其他用户,推荐这些用户喜欢的物品。

import numpy as np
import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity
from scipy.sparse import csr_matrix
class UserBasedCF:
    def __init__(self, k_neighbors=5):
        self.k_neighbors = k_neighbors
        self.user_similarity = None
        self.user_item_matrix = None
        self.user_mapping = {}
        self.item_mapping = {}
    def fit(self, ratings_df):
        """
        ratings_df: DataFrame with columns ['user_id', 'item_id', 'rating']
        """
        # 创建用户-物品矩阵
        self.user_item_matrix = ratings_df.pivot_table(
            index='user_id', 
            columns='item_id', 
            values='rating'
        ).fillna(0)
        # 计算用户之间的余弦相似度
        self.user_similarity = cosine_similarity(self.user_item_matrix)
        self.user_similarity = pd.DataFrame(
            self.user_similarity,
            index=self.user_item_matrix.index,
            columns=self.user_item_matrix.index
        )
    def predict_rating(self, user_id, item_id):
        """
        预测用户对物品的评分
        """
        if user_id not in self.user_similarity.index or item_id not in self.user_item_matrix.columns:
            return 0
        # 获取相似用户
        similar_users = self.user_similarity[user_id].drop(user_id).sort_values(ascending=False)
        # 选择前k个相似用户
        top_k_users = similar_users.head(self.k_neighbors)
        # 计算加权平均评分
        numerator = 0
        denominator = 0
        for sim_user, similarity in top_k_users.items():
            rating = self.user_item_matrix.loc[sim_user, item_id]
            if rating > 0:  # 用户对这个物品有过评分
                numerator += similarity * rating
                denominator += similarity
        if denominator == 0:
            return 0
        return numerator / denominator
    def recommend(self, user_id, n_items=5):
        """
        为用户推荐物品
        """
        # 获取用户未评分的物品
        user_rated = self.user_item_matrix.loc[user_id]
        unrated_items = user_rated[user_rated == 0].index
        # 预测未评分物品的评分
        predictions = []
        for item_id in unrated_items:
            pred_rating = self.predict_rating(user_id, item_id)
            predictions.append((item_id, pred_rating))
        # 排序并返回top-n推荐
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:n_items]

基于物品的协同过滤

核心思想

找��与目标物品相似的物品,推荐这些物品。

class ItemBasedCF:
    def __init__(self, k_neighbors=5):
        self.k_neighbors = k_neighbors
        self.item_similarity = None
        self.user_item_matrix = None
    def fit(self, ratings_df):
        """
        ratings_df: DataFrame with columns ['user_id', 'item_id', 'rating']
        """
        # 创建用户-物品矩阵（转置）
        self.user_item_matrix = ratings_df.pivot_table(
            index='user_id', 
            columns='item_id', 
            values='rating'
        ).fillna(0)
        # 计算物品之间的相似度（转置矩阵）
        item_matrix = self.user_item_matrix.T
        self.item_similarity = cosine_similarity(item_matrix)
        self.item_similarity = pd.DataFrame(
            self.item_similarity,
            index=item_matrix.index,
            columns=item_matrix.index
        )
    def predict_rating(self, user_id, item_id):
        """
        预测用户对物品的评分
        """
        if user_id not in self.user_item_matrix.index:
            return 0
        # 获取用户评分过的物品
        user_ratings = self.user_item_matrix.loc[user_id]
        # 计算加权平均相似度
        numerator = 0
        denominator = 0
        for rated_item, rating in user_ratings.items():
            if rating > 0 and rated_item != item_id:
                if item_id in self.item_similarity.columns:
                    similarity = self.item_similarity.loc[item_id, rated_item]
                    numerator += similarity * rating
                    denominator += abs(similarity)
        if denominator == 0:
            return 0
        return numerator / denominator
    def recommend(self, user_id, n_items=5):
        """
        为用户推荐物品
        """
        user_rated = self.user_item_matrix.loc[user_id]
        unrated_items = user_rated[user_rated == 0].index
        predictions = []
        for item_id in unrated_items:
            pred_rating = self.predict_rating(user_id, item_id)
            predictions.append((item_id, pred_rating))
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:n_items]

矩阵分解方法（SVD）

核心思想

将用户-物品矩阵分解为低维特征矩阵。

from sklearn.decomposition import TruncatedSVD
from scipy.sparse import csr_matrix
class MatrixFactorization:
    def __init__(self, n_factors=20):
        self.n_factors = n_factors
        self.svd = None
        self.user_item_matrix = None
        self.user_mapping = {}
        self.item_mapping = {}
    def fit(self, ratings_df):
        """
        使用SVD进行矩阵分解
        """
        # 创建用户-物品矩阵
        self.user_item_matrix = ratings_df.pivot_table(
            index='user_id', 
            columns='item_id', 
            values='rating'
        ).fillna(0)
        # 稀疏矩阵化
        sparse_matrix = csr_matrix(self.user_item_matrix.values)
        # SVD分解
        self.svd = TruncatedSVD(n_components=self.n_factors, random_state=42)
        self.latent_features = self.svd.fit_transform(sparse_matrix)
        self.item_latent = self.svd.components_.T
    def predict_rating(self, user_idx, item_idx):
        """
        基于矩阵分解预测评分
        """
        if user_idx < len(self.latent_features) and item_idx < len(self.item_latent):
            return np.dot(self.latent_features[user_idx], self.item_latent[item_idx])
        return 0
    def recommend(self, user_id, n_items=5):
        """
        为用户推荐物品
        """
        # 获取用户索引
        user_idx = list(self.user_item_matrix.index).index(user_id)
        # 预测所有未评分物品的评分
        predictions = []
        for item_idx in range(len(self.user_item_matrix.columns)):
            if self.user_item_matrix.iloc[user_idx, item_idx] == 0:
                pred = self.predict_rating(user_idx, item_idx)
                item_id = self.user_item_matrix.columns[item_idx]
                predictions.append((item_id, pred))
        # 排序并返回top-n
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:n_items]

使用Surprise库的简单实现

from surprise import Dataset, Reader, SVD, KNNBasic
from surprise.model_selection import train_test_split
from surprise import accuracy
class SurpriseCF:
    def __init__(self, algorithm='SVD'):
        self.algorithm = algorithm
        self.model = None
    def fit(self, ratings_df):
        """
        使用Surprise库实现协同过滤
        """
        # 准备数据
        reader = Reader(rating_scale=(1, 5))
        data = Dataset.load_from_df(ratings_df[['user_id', 'item_id', 'rating']], reader)
        # 划分训练集和测试集
        trainset, testset = train_test_split(data, test_size=0.2, random_state=42)
        # 选择算法
        if self.algorithm == 'SVD':
            self.model = SVD(n_factors=20, random_state=42)
        elif self.algorithm == 'KNN':
            self.model = KNNBasic(k=20, sim_options={'name': 'cosine', 'user_based': True})
        # 训练模型
        self.model.fit(trainset)
        # 评估模型
        predictions = self.model.test(testset)
        rmse = accuracy.rmse(predictions)
        print(f"RMSE: {rmse}")
    def predict(self, user_id, item_id):
        """
        预测评分
        """
        return self.model.predict(user_id, item_id)
    def recommend(self, user_id, n_items=5):
        """
        为用户推荐物品
        """
        # 获取所有物品
        all_items = self.model.trainset.all_items()
        user_items = self.model.trainset.ur[self.model.trainset.to_inner_uid(user_id)]
        rated_items = set([item[0] for item in user_items])
        # 预测未评分物品的评分
        predictions = []
        for inner_item_id in all_items:
            if inner_item_id not in rated_items:
                item_id = self.model.trainset.to_raw_iid(inner_item_id)
                pred = self.model.predict(user_id, item_id)
                predictions.append((item_id, pred.est))
        # 排序并返回top-n
        predictions.sort(key=lambda x: x[1], reverse=True)
        return predictions[:n_items]

使用示例

# 示例数据
ratings_data = {
    'user_id': [1, 1, 1, 2, 2, 3, 3, 3, 4, 4],
    'item_id': [1, 2, 3, 1, 4, 2, 3, 4, 1, 5],
    'rating': [5, 3, 4, 4, 2, 5, 3, 4, 3, 5]
}
ratings_df = pd.DataFrame(ratings_data)
# 使用基于用户的协同过滤
user_cf = UserBasedCF(k_neighbors=3)
user_cf.fit(ratings_df)
recommendations = user_cf.recommend(1, n_items=3)
print("User-based CF recommendations for user 1:", recommendations)
# 使用矩阵分解
mf = MatrixFactorization(n_factors=10)
mf.fit(ratings_df)
recommendations = mf.recommend(1, n_items=3)
print("Matrix Factorization recommendations for user 1:", recommendations)

优化建议

class OptimizedCF:
    def __init__(self):
        self.cache = {}  # 缓存计算结果
    def calculate_similarity_with_cache(self, user_id1, user_id2):
        """缓存相似度计算结果"""
        key = f"{min(user_id1, user_id2)}_{max(user_id1, user_id2)}"
        if key not in self.cache:
            # 计算相似度...
            self.cache[key] = similarity
        return self.cache[key]
    def recommend_with_batch_processing(self, user_id, n_items=5):
        """批量处理推荐"""
        # 使用向量化操作替代循环
        pass

基于用户：适合物品数量多但用户少的情况
基于物品：适合用户数量多但物品少的情况
矩阵分解：准确度较高，适合大规模数据
混合方法：结合多种方法获得更好效果

实际应用中需要根据数据规模、实时性要求、准确度需求选择合适的算法。