如何用Python快速构建自动化推荐系统脚本？

推荐系统的落地往往不需要从零实现复杂的矩阵分解或深度学习模型，尤其是当目标是自动化输出个性化列表时，基于用户行为数据的协同过滤就能满足多数场景。Python生态中的pandas和scikit-learn提供了足够高效的矩阵运算能力，配合schedule等调度库，可以把推荐计算变成无人值守的定时任务。本文以用户对物品的显式评分为例，完整演示从数据准备到脚本部署的过程。

数据准备：从原始记录到评分矩阵

实际业务中，用户行为通常记录为三列：用户ID、物品ID、评分（或点击、购买等隐式行为的权重）。如果直接在这些长表上做相似度计算，不仅效率低下，而且逻辑会变得混乱。第一步需要把记录转换成用户-物品的二维矩阵，行代表用户，列代表物品，单元格是评分值。

下面这段代码演示了如何用pandas的pivot_table方法完成转换。缺失值用0填充，表示该用户尚未评价过对应物品。这种处理方式简单直观，但会引入一个问题：评分矩阵会变得非常稀疏，并且0被当成已知的低分处理。后续计算相似度时，我们只关注共同评分过的物品，避免0值影响判断。

import pandas as pd
from sklearn.metrics.pairwise import cosine_similarity

# 原始用户评分记录
ratings = pd.DataFrame({
    'user_id': [1, 1, 1, 2, 2, 3, 3, 3, 4],
    'item_id': [101, 102, 103, 101, 104, 102, 103, 105, 101],
    'rating': [5, 3, 4, 4, 2, 5, 5, 1, 3]
})

# 长表转宽表，构建用户-物品评分矩阵
rating_matrix = ratings.pivot_table(
    index='user_id',
    columns='item_id',
    values='rating'
).fillna(0)

print(rating_matrix.head())

如果你处理的是隐式反馈数据，比如点击、浏览、收藏，可以把行为次数或二值化后的1作为评分。需要注意的是，评分矩阵的规模会随用户数和物品数快速增长，对于百万级数据，建议直接使用稀疏矩阵存储，例如scipy.sparse中的CSR格式，而不是密集的DataFrame。

相似度计算与推荐生成

协同过滤的关键在于找到相似的用户或物品。用户相似度通常用余弦相似度计算，公式是两用户共同评分向量夹角的余弦值。scikit-learn的cosine_similarity函数底层做了优化，可以直接传入评分矩阵，返回对称的相似度矩阵。

得到相似度后，就可以为每个用户生成推荐列表。对于某个用户未评分的物品，找出所有评价过该物品的其他用户，用他们与该用户的相似度作为权重，对评分进行加权平均，得到预测分数。这个分数越高，说明用户越可能喜欢该物品。下面这个函数实现了完整的逻辑，并返回Top-N推荐。

def recommend_for_user(user_id, rating_matrix, user_sim_df, top_n=5):
    if user_id not in rating_matrix.index:
        return []
    # 当前用户已评分的物品
    user_rated = rating_matrix.loc[user_id]
    unrated_items = user_rated[user_rated == 0].index
    score_map = {}
    for item in unrated_items:
        # 找出评价过该物品的其他用户
        rated_users = rating_matrix[item][rating_matrix[item] > 0].index
        weighted_sum = 0.0
        sim_sum = 0.0
        for other_user in rated_users:
            if other_user == user_id:
                continue
            similarity = user_sim_df.loc[user_id, other_user]
            rating_value = rating_matrix.loc[other_user, item]
            weighted_sum += similarity * rating_value
            sim_sum += abs(similarity)
        if sim_sum > 0:
            score_map[item] = weighted_sum / sim_sum
    sorted_items = sorted(score_map.items(), key=lambda kv: kv[1], reverse=True)
    return [item for item, score in sorted_items[:top_n]]

这个函数的时间复杂度与用户数和物品数相关，在数据量不大时完全够用。如果物品数量较多，建议改为物品相似度：预先计算物品两两相似度，再基于用户已评分的物品累加相似物品的分数。物品数量通常远小于用户数时，物品相似度计算量更小，推荐结果也更稳定。

封装成自动化脚本

单次计算的代码只适合在Notebook里调试。要把推荐系统变成自动化流程，需要解决三个问题：输入输出的参数化、运行日志和异常告警、定时调度。

输入输出参数化可以用argparse或简单的函数参数。下面这个脚本将数据读取、矩阵构建、推荐生成和结果保存封装成一个可重复调用的函数，并在每天凌晨2点自动执行。如果数据文件不存在或格式错误，try/except会捕获异常并记录日志，而不是直接中断整个任务。

import schedule
import time
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def run_recommendation_job(data_path, output_path, top_n=5):
    logging.info('开始读取数据 %s', data_path)
    ratings = pd.read_csv(data_path)
    rating_matrix = ratings.pivot_table(index='user_id', columns='item_id', values='rating').fillna(0)
    user_similarity = cosine_similarity(rating_matrix)
    user_sim_df = pd.DataFrame(user_similarity, index=rating_matrix.index, columns=rating_matrix.index)

    all_recs = []
    for user_id in rating_matrix.index:
        rec_items = recommend_for_user(user_id, rating_matrix, user_sim_df, top_n)
        for rank, item in enumerate(rec_items, start=1):
            all_recs.append({'user_id': user_id, 'item_id': item, 'rank': rank})
    rec_df = pd.DataFrame(all_recs)
    rec_df.to_csv(output_path, index=False)
    logging.info('推荐结果已保存至 %s', output_path)

# 每天凌晨2点执行任务
schedule.every().day.at('02:00').do(
    run_recommendation_job,
    data_path='ratings.csv',
    output_path='recommendations.csv'
)

while True:
    schedule.run_pending()
    time.sleep(60)

生产环境中不建议把调度器写在Python进程内部一直循环，可以改用Linux的cron或Windows的任务计划程序，通过命令行调用脚本。例如执行python rec_job.py --data ratings.csv --output recs.csv。这样进程在执行完任务后自动退出，不会常驻内存。日志可以输出到文件，方便后续排查。

冷启动与性能优化技巧

冷启动是推荐系统绕不开的问题。新用户没有任何评分记录，基于用户的协同过滤无法给出可靠的个性化推荐。一个常用的兜底方案是直接返回全局热门物品，或者根据用户注册时选择的兴趣标签做基于内容的推荐。你可以在recommend_for_user函数开头加一个判断：如果用户未在评分矩阵中出现，则返回热门列表。

性能方面，对于大规模数据，全量计算用户相似度矩阵的时间复杂度是O(n^2 * m)，n是用户数，m是物品数，用户数增长时计算量会急剧上升。可以改用sklearn.neighbors.NearestNeighbors只寻找每个用户的K个最近邻，或者使用surprise库内置的KNNBasic算法，它底层使用稀疏矩阵和近似搜索，速度更快。另外，如果数据更新频率不高，可以把相似度矩阵缓存到磁盘，增量更新时只重新计算发生变化的部分。

从工程角度看，推荐脚本的健壮性比算法精度更重要。为每次运行设置超时时间，为相似度计算添加降级方案，例如在数据量过大时自动切换为随机推荐或热门推荐，并记录告警信息。这样即使出现预料之外的数据波动，也能保证服务不会完全不可用。

本文给出的代码已经涵盖了自动化推荐脚本的核心骨架，你可以根据实际业务需求调整评分字段、相似度算法和推荐数量，快速在项目中落地。

Python推荐系统自动化脚本协同过滤修改时间：2026-09-23 10:54:41