用Python从零搭建电影推荐系统：MovieLens-100K数据集实战

推荐系统已经成为数字时代不可或缺的技术支柱。想象一下，当你打开一个流媒体平台，首页上那些恰好符合你口味的电影推荐，背后就是推荐算法在发挥作用。对于Python开发者来说，亲手构建一个推荐系统不仅能加深对算法的理解，还能掌握从数据处理到模型部署的完整流程。本文将带你用MovieLens-100K数据集，从零开始实现四种经典推荐算法，并附上可直接运行的完整代码。

1. 环境准备与数据加载

1.1 搭建Python环境

推荐系统开发需要一些核心Python库的支持。建议使用conda或virtualenv创建隔离的Python环境：

conda create -n recsys python=3.9
conda activate recsys
pip install numpy pandas scikit-learn scipy matplotlib

这些库将帮助我们完成：

• numpy：高效的数值计算
• pandas：数据处理与分析
• scikit-learn：机器学习算法实现
• scipy：科学计算工具
• matplotlib：数据可视化

1.2 获取MovieLens-100K数据集

MovieLens数据集是推荐系统研究的黄金标准。100K版本包含943位用户对1682部电影的100,000条评分（1-5分）。下载数据集：

import os
import pandas as pd
from urllib.request import urlretrieve

# 创建数据目录
if not os.path.exists('ml-100k'):
    os.makedirs('ml-100k')

# 下载数据集文件
files = {
    'u.data': 'https://files.grouplens.org/datasets/movielens/ml-100k/u.data',
    'u.item': 'https://files.grouplens.org/datasets/movielens/ml-100k/u.item',
    'u.user': 'https://files.grouplens.org/datasets/movielens/ml-100k/u.user'
}

for filename, url in files.items():
    if not os.path.exists(f'ml-100k/{filename}'):
        urlretrieve(url, f'ml-100k/{filename}')
        print(f'已下载: {filename}')

1.3 数据加载与初步探索

加载三个核心数据文件：

# 加载评分数据
ratings = pd.read_csv(
    'ml-100k/u.data', 
    sep='\t', 
    names=['user_id', 'item_id', 'rating', 'timestamp'],
    encoding='latin-1'
)

# 加载电影数据
movies = pd.read_csv(
    'ml-100k/u.item', 
    sep='|', 
    names=['item_id', 'title', 'release_date', 'video_release_date', 
           'IMDb_URL', 'unknown', 'Action', 'Adventure', 'Animation', 
           'Children', 'Comedy', 'Crime', 'Documentary', 'Drama', 
           'Fantasy', 'Film-Noir', 'Horror', 'Musical', 'Mystery', 
           'Romance', 'Sci-Fi', 'Thriller', 'War', 'Western'],
    encoding='latin-1',
    usecols=['item_id', 'title']
)

# 加载用户数据
users = pd.read_csv(
    'ml-100k/u.user', 
    sep='|', 
    names=['user_id', 'age', 'gender', 'occupation', 'zip_code'],
    encoding='latin-1'
)

查看数据集基本信息：

print(f"用户数量: {users['user_id'].nunique()}")
print(f"电影数量: {movies['item_id'].nunique()}")
print(f"评分记录: {len(ratings)}")
print(f"评分范围: {ratings['rating'].min()} - {ratings['rating'].max()}")
print(f"数据稀疏度: {len(ratings)/(943*1682):.4f}")  # 943用户×1682电影

2. 数据预处理与特征工程

2.1 构建用户-电影评分矩阵

推荐系统的核心是用户-物品交互矩阵。我们将其转换为稀疏矩阵形式：

from scipy.sparse import csr_matrix

# 创建用户-电影评分矩阵
user_movie_matrix = ratings.pivot_table(
    index='user_id',
    columns='item_id',
    values='rating',
    fill_value=0
)

# 确保包含所有用户和电影
all_users = range(1, 944)  # 用户ID从1到943
all_movies = range(1, 1683)  # 电影ID从1到1682
user_movie_matrix = user_movie_matrix.reindex(
    index=all_users,
    columns=all_movies,
    fill_value=0
)

# 转换为稀疏矩阵节省内存
sparse_matrix = csr_matrix(user_movie_matrix.values)

2.2 数据标准化与归一化

不同用户的评分习惯差异很大，有的用户倾向于打高分，有的则比较严格。我们需要进行标准化处理：

from sklearn.preprocessing import normalize

# 按行(用户)标准化
normalized_matrix = normalize(sparse_matrix, norm='l2', axis=1)

2.3 划分训练集与测试集

为了评估模型性能，我们需要将数据分为训练集和测试集：

from sklearn.model_selection import train_test_split

# 转换为coo格式便于拆分
coo_matrix = sparse_matrix.tocoo()

# 获取非零元素的行列索引和值
rows = coo_matrix.row
cols = coo_matrix.col
data = coo_matrix.data

# 划分训练测试集
train_rows, test_rows, train_cols, test_cols, train_data, test_data = train_test_split(
    rows, cols, data, test_size=0.2, random_state=42
)

# 重建训练和测试矩阵
train_matrix = csr_matrix((train_data, (train_rows, train_cols)), shape=sparse_matrix.shape)
test_matrix = csr_matrix((test_data, (test_rows, test_cols)), shape=sparse_matrix.shape)

3. 实现协同过滤推荐算法

3.1 基于用户的协同过滤(User-CF)

User-CF的核心思想是"相似用户喜欢相似物品"。实现步骤如下：

1. 计算用户相似度矩阵
2. 为每个用户找到K个最相似用户
3. 基于相似用户的评分预测目标用户对未评分物品的评分

from sklearn.metrics.pairwise import cosine_similarity

def user_based_cf(user_id, train_matrix, k=20):
    """基于用户的协同过滤推荐"""
    # 计算用户相似度
    user_sim = cosine_similarity(train_matrix)
    
    # 获取目标用户的相似用户(排除自己)
    sim_users = np.argsort(-user_sim[user_id-1])[1:k+1]
    
    # 获取目标用户未评分的电影
    rated_items = set(train_matrix[user_id-1].indices)
    all_items = set(range(train_matrix.shape[1]))
    candidate_items = all_items - rated_items
    
    # 预测评分
    predictions = {}
    for item in candidate_items:
        # 获取相似用户对该物品的评分
        ratings = train_matrix[sim_users, item].toarray().flatten()
        # 获取相似度权重
        weights = user_sim[user_id-1, sim_users]
        # 计算加权平均评分
        if np.sum(weights[ratings > 0]) > 0:
            pred = np.dot(weights, ratings) / np.sum(weights[ratings > 0])
            predictions[item] = pred
    
    # 返回按预测评分排序的结果
    return sorted(predictions.items(), key=lambda x: x[1], reverse=True)

3.2 基于物品的协同过滤(Item-CF)

Item-CF的核心思想是"用户喜欢与他们过去喜欢的物品相似的物品"。实现步骤：

1. 计算物品相似度矩阵
2. 为用户已评分物品找到K个最相似物品
3. 基于相似度和用户历史评分预测目标用户对未评分物品的评分

def item_based_cf(user_id, train_matrix, k=20):
    """基于物品的协同过滤推荐"""
    # 计算物品相似度(转置矩阵)
    item_sim = cosine_similarity(train_matrix.T)
    
    # 获取用户已评分且评分≥3的物品(视为用户喜欢的物品)
    user_ratings = train_matrix[user_id-1]
    liked_items = [i for i in user_ratings.indices if user_ratings[0, i] >= 3]
    
    # 获取用户未评分的电影
    rated_items = set(user_ratings.indices)
    all_items = set(range(train_matrix.shape[1]))
    candidate_items = all_items - rated_items
    
    # 预测评分
    predictions = {}
    for item in candidate_items:
        # 获取与目标物品最相似的K个已评分物品
        sim_items = np.argsort(-item_sim[item])[:k]
        # 只考虑用户实际评分过的相似物品
        valid_sim_items = [i for i in sim_items if i in liked_items]
        if valid_sim_items:
            # 获取用户对这些相似物品的评分
            ratings = [user_ratings[0, i] for i in valid_sim_items]
            # 获取相似度权重
            weights = [item_sim[item, i] for i in valid_sim_items]
            # 计算加权平均评分
            pred = np.dot(weights, ratings) / np.sum(weights)
            predictions[item] = pred
    
    # 返回按预测评分排序的结果
    return sorted(predictions.items(), key=lambda x: x[1], reverse=True)

4. 实现矩阵分解推荐算法

4.1 奇异值分解(SVD)

SVD通过将评分矩阵分解为低维用户特征和物品特征矩阵来捕捉潜在因素：

from sklearn.decomposition import TruncatedSVD

def svd_recommend(user_id, train_matrix, n_components=50):
    """基于SVD的推荐"""
    # 训练SVD模型
    svd = TruncatedSVD(n_components=n_components, random_state=42)
    user_features = svd.fit_transform(train_matrix)
    item_features = svd.components_
    
    # 重建评分矩阵
    reconstructed = np.dot(user_features, item_features)
    
    # 获取用户未评分的电影
    rated_items = set(train_matrix[user_id-1].indices)
    all_items = set(range(train_matrix.shape[1]))
    candidate_items = all_items - rated_items
    
    # 生成预测评分
    predictions = [(item, reconstructed[user_id-1, item]) 
                   for item in candidate_items]
    
    # 返回按预测评分排序的结果
    return sorted(predictions, key=lambda x: x[1], reverse=True)

4.2 隐语义模型(LFM)

LFM通过优化目标函数学习用户和物品的潜在特征：

class LatentFactorModel:
    def __init__(self, n_factors=50, lr=0.01, reg=0.02, n_epochs=50):
        self.n_factors = n_factors  # 潜在因子数量
        self.lr = lr  # 学习率
        self.reg = reg  # 正则化系数
        self.n_epochs = n_epochs  # 训练轮数
        
    def fit(self, train_matrix):
        """训练LFM模型"""
        n_users, n_items = train_matrix.shape
        
        # 初始化参数
        self.user_factors = np.random.normal(scale=1./self.n_factors, 
                                           size=(n_users, self.n_factors))
        self.item_factors = np.random.normal(scale=1./self.n_factors,
                                           size=(n_items, self.n_factors))
        self.user_bias = np.zeros(n_users)
        self.item_bias = np.zeros(n_items)
        self.global_bias = np.mean(train_matrix.data)
        
        # 转换为coo格式便于迭代
        coo = train_matrix.tocoo()
        
        # 训练过程
        for epoch in range(self.n_epochs):
            for u, i, r in zip(coo.row, coo.col, coo.data):
                # 计算预测误差
                prediction = self.global_bias + self.user_bias[u] + self.item_bias[i]
                prediction += np.dot(self.user_factors[u], self.item_factors[i])
                error = r - prediction
                
                # 更新参数
                self.user_bias[u] += self.lr * (error - self.reg * self.user_bias[u])
                self.item_bias[i] += self.lr * (error - self.reg * self.item_bias[i])
                
                # 更新因子
                u_factor = self.user_factors[u]
                i_factor = self.item_factors[i]
                self.user_factors[u] += self.lr * (error * i_factor - self.reg * u_factor)
                self.item_factors[i] += self.lr * (error * u_factor - self.reg * i_factor)
    
    def predict(self, user_id, item_id):
        """预测用户对物品的评分"""
        pred = self.global_bias + self.user_bias[user_id-1] + self.item_bias[item_id-1]
        pred += np.dot(self.user_factors[user_id-1], self.item_factors[item_id-1])
        return max(1, min(5, pred))  # 确保评分在1-5范围内

def lfm_recommend(user_id, train_matrix, model):
    """基于LFM的推荐"""
    # 获取用户未评分的电影
    rated_items = set(train_matrix[user_id-1].indices)
    all_items = set(range(train_matrix.shape[1]))
    candidate_items = all_items - rated_items
    
    # 生成预测评分
    predictions = [(item, model.predict(user_id, item+1)) 
                   for item in candidate_items]
    
    # 返回按预测评分排序的结果
    return sorted(predictions, key=lambda x: x[1], reverse=True)

5. 模型评估与结果分析

5.1 评估指标实现

我们使用MAE(平均绝对误差)和RMSE(均方根误差)评估预测准确性：

from sklearn.metrics import mean_absolute_error, mean_squared_error

def evaluate(predictions, test_matrix):
    """评估模型在测试集上的表现"""
    # 获取测试集中的真实评分
    test_coo = test_matrix.tocoo()
    true_ratings = test_coo.data
    user_indices = test_coo.row
    item_indices = test_coo.col
    
    # 获取预测评分
    pred_ratings = []
    for u, i in zip(user_indices, item_indices):
        pred = predictions[u][i] if i in predictions[u] else 0
        pred_ratings.append(pred)
    
    # 计算指标
    mae = mean_absolute_error(true_ratings, pred_ratings)
    rmse = np.sqrt(mean_squared_error(true_ratings, pred_ratings))
    
    return mae, rmse

5.2 各算法性能对比

训练并评估四种算法：

# 训练LFM模型
lfm = LatentFactorModel(n_factors=50, lr=0.01, reg=0.02, n_epochs=50)
lfm.fit(train_matrix)

# 评估各算法
algorithms = {
    'User-CF': user_based_cf,
    'Item-CF': item_based_cf,
    'SVD': svd_recommend,
    'LFM': lambda u, m: lfm_recommend(u, m, lfm)
}

results = {}
for name, algo in algorithms.items():
    # 为测试集中每个用户生成推荐
    predictions = {}
    for u in range(train_matrix.shape[0]):
        recs = algo(u+1, train_matrix)
        predictions[u] = {i: r for i, r in recs}
    
    # 评估
    mae, rmse = evaluate(predictions, test_matrix)
    results[name] = {'MAE': mae, 'RMSE': rmse}
    print(f"{name}: MAE={mae:.4f}, RMSE={rmse:.4f}")

典型评估结果可能如下：

算法	MAE	RMSE
User-CF	0.8437	1.1129
Item-CF	0.8483	1.0694
SVD	1.4216	1.7533
LFM	0.9215	1.1873

5.3 为特定用户生成推荐

让我们为用户ID=4生成个性化推荐：

def get_recommendations(user_id, algorithms, train_matrix, movies, top_n=5):
    """获取各算法的推荐结果"""
    recommendations = {}
    for name, algo in algorithms.items():
        if name == 'LFM':
            recs = algo(user_id, train_matrix, lfm)
        else:
            recs = algo(user_id, train_matrix)
        
        # 获取电影标题
        top_recs = []
        for item_id, pred_rating in recs[:top_n]:
            title = movies[movies['item_id'] == item_id+1]['title'].iloc[0]
            top_recs.append({'title': title, 'pred_rating': round(pred_rating, 2)})
        
        recommendations[name] = top_recs
    
    return recommendations

# 获取用户4的历史高评分电影
user_rated = ratings[ratings['user_id'] == 4]
high_rated = user_rated[user_rated['rating'] >= 4].merge(
    movies, on='item_id'
).sort_values('rating', ascending=False)

print("用户4的历史高评分电影:")
print(high_rated[['title', 'rating']].head())

# 生成推荐
recs = get_recommendations(4, algorithms, train_matrix, movies)
for algo, items in recs.items():
    print(f"\n{algo}推荐结果:")
    for i, item in enumerate(items, 1):
        print(f"{i}. {item['title']} (预测评分: {item['pred_rating']})")

6. 系统优化与扩展方向

6.1 冷启动问题解决方案

新用户或新物品缺乏足够交互数据时，推荐质量会显著下降。解决方案包括：

1. 混合推荐：结合基于内容的推荐(使用物品元数据)
2. 随机探索：定期向用户推荐随机物品收集反馈
3. 利用人口统计信息：对新用户使用年龄、性别等特征

6.2 实时推荐实现

传统批处理推荐延迟高，实时推荐能更快响应用户行为：

from collections import defaultdict

class RealTimeRecommender:
    def __init__(self, n_factors=50):
        self.user_factors = defaultdict(lambda: np.random.normal(size=n_factors))
        self.item_factors = defaultdict(lambda: np.random.normal(size=n_factors))
        self.global_bias = 3.0  # 初始全局平均评分
    
    def update(self, user_id, item_id, rating, lr=0.01):
        """根据新评分实时更新模型"""
        # 计算预测误差
        pred = self.global_bias + np.dot(self.user_factors[user_id], self.item_factors[item_id])
        error = rating - pred
        
        # 更新参数
        u_factor = self.user_factors[user_id]
        i_factor = self.item_factors[item_id]
        
        self.user_factors[user_id] += lr * (error * i_factor - 0.01 * u_factor)
        self.item_factors[item_id] += lr * (error * u_factor - 0.01 * i_factor)
        self.global_bias += lr * error
    
    def recommend(self, user_id, candidate_items, top_n=5):
        """生成实时推荐"""
        scores = [(item, np.dot(self.user_factors[user_id], self.item_factors[item]))
                 for item in candidate_items]
        return sorted(scores, key=lambda x: x[1], reverse=True)[:top_n]

6.3 部署为Web服务

使用Flask将推荐系统部署为REST API：

from flask import Flask, request, jsonify
import pickle

app = Flask(__name__)

# 加载预训练模型
with open('recommender.pkl', 'rb') as f:
    model = pickle.load(f)

@app.route('/recommend', methods=['POST'])
def recommend():
    data = request.json
    user_id = data['user_id']
    candidate_items = data['candidate_items']
    
    # 获取推荐
    recs = model.recommend(user_id, candidate_items)
    
    return jsonify({
        'user_id': user_id,
        'recommendations': recs
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

7. 完整代码整合与使用指南

将所有组件整合为一个完整的推荐系统类：

import numpy as np
import pandas as pd
from scipy.sparse import csr_matrix
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.decomposition import TruncatedSVD
from sklearn.model_selection import train_test_split

class MovieRecommender:
    def __init__(self):
        self.ratings = None
        self.movies = None
        self.users = None
        self.train_matrix = None
        self.test_matrix = None
        self.user_sim = None
        self.item_sim = None
        self.svd_model = None
        self.lfm_model = None
    
    def load_data(self, data_dir='ml-100k'):
        """加载MovieLens-100K数据集"""
        # 加载评分数据
        self.ratings = pd.read_csv(
            f'{data_dir}/u.data', 
            sep='\t', 
            names=['user_id', 'item_id', 'rating', 'timestamp'],
            encoding='latin-1'
        )
        
        # 加载电影数据
        self.movies = pd.read_csv(
            f'{data_dir}/u.item', 
            sep='|', 
            names=['item_id', 'title', 'release_date', 'video_release_date', 
                   'IMDb_URL', 'unknown', 'Action', 'Adventure', 'Animation', 
                   'Children', 'Comedy', 'Crime', 'Documentary', 'Drama', 
                   'Fantasy', 'Film-Noir', 'Horror', 'Musical', 'Mystery', 
                   'Romance', 'Sci-Fi', 'Thriller', 'War', 'Western'],
            encoding='latin-1',
            usecols=['item_id', 'title']
        )
        
        # 加载用户数据
        self.users = pd.read_csv(
            f'{data_dir}/u.user', 
            sep='|', 
            names=['user_id', 'age', 'gender', 'occupation', 'zip_code'],
            encoding='latin-1'
        )
    
    def prepare_matrix(self, test_size=0.2):
        """准备训练和测试矩阵"""
        # 创建用户-电影评分矩阵
        user_movie_matrix = self.ratings.pivot_table(
            index='user_id',
            columns='item_id',
            values='rating',
            fill_value=0
        )
        
        # 确保包含所有用户和电影
        all_users = range(1, 944)
        all_movies = range(1, 1683)
        user_movie_matrix = user_movie_matrix.reindex(
            index=all_users,
            columns=all_movies,
            fill_value=0
        )
        
        # 转换为稀疏矩阵
        sparse_matrix = csr_matrix(user_movie_matrix.values)
        
        # 划分训练测试集
        coo_matrix = sparse_matrix.tocoo()
        rows = coo_matrix.row
        cols = coo_matrix.col
        data = coo_matrix.data
        
        train_rows, test_rows, train_cols, test_cols, train_data, test_data = train_test_split(
            rows, cols, data, test_size=test_size, random_state=42
        )
        
        self.train_matrix = csr_matrix((train_data, (train_rows, train_cols)), 
                                     shape=sparse_matrix.shape)
        self.test_matrix = csr_matrix((test_data, (test_rows, test_cols)), 
                                    shape=sparse_matrix.shape)
    
    def train_models(self):
        """训练所有推荐模型"""
        # 计算用户相似度(User-CF)
        self.user_sim = cosine_similarity(self.train_matrix)
        
        # 计算物品相似度(Item-CF)
        self.item_sim = cosine_similarity(self.train_matrix.T)
        
        # 训练SVD模型
        self.svd_model = TruncatedSVD(n_components=50, random_state=42)
        self.svd_model.fit(self.train_matrix)
        
        # 训练LFM模型
        self.lfm_model = LatentFactorModel()
        self.lfm_model.fit(self.train_matrix)
    
    def recommend(self, user_id, algo='user_cf', top_n=5):
        """生成推荐"""
        if algo == 'user_cf':
            recs = user_based_cf(user_id, self.train_matrix)
        elif algo == 'item_cf':
            recs = item_based_cf(user_id, self.train_matrix)
        elif algo == 'svd':
            recs = svd_recommend(user_id, self.train_matrix)
        elif algo == 'lfm':
            recs = lfm_recommend(user_id, self.train_matrix, self.lfm_model)
        else:
            raise ValueError("不支持的算法类型")
        
        # 获取电影标题
        recommendations = []
        for item_id, pred_rating in recs[:top_n]:
            title = self.movies[self.movies['item_id'] == item_id+1]['title'].iloc[0]
            recommendations.append({
                'item_id': item_id+1,
                'title': title,
                'pred_rating': round(pred_rating, 2)
            })
        
        return recommendations

使用示例：

# 初始化推荐系统
recommender = MovieRecommender()

# 加载数据
recommender.load_data()

# 准备矩阵
recommender.prepare_matrix()

# 训练模型
recommender.train_models()

# 为用户4生成推荐
user_id = 4
print(f"\n为用户 {user_id} 生成的推荐:")

# User-CF推荐
user_cf_recs = recommender.recommend(user_id, 'user_cf')
print("\nUser-CF推荐:")
for rec in user_cf_recs:
    print(f"{rec['item_id']}\t{rec['title']}\t(预测评分: {rec['pred_rating']})")

# Item-CF推荐
item_cf_recs = recommender.recommend(user_id, 'item_cf')
print("\nItem-CF推荐:")
for rec in item_cf_recs:
    print(f"{rec['item_id']}\t{rec['title']}\t(预测评分: {rec['pred_rating']})")