基于SVD协同过滤与内容过滤的酒庄推荐系统：从数据清洗到FastAPI落地

简介：基于Python的酒庄数据分析推荐系统项目实例，面向具备一定Python编程基础、熟悉数据分析与Web开发的研发人员、数据科学家或软件工程师，以数据驱动方式解决酒庄线上商城与线下门店的个性化推荐、会员运营和库存优化等实际问题。项目覆盖需求分析、MySQL数据库设计、Flask后端接口、Tkinter图形界面及推荐算法实现，采用协同过滤与内容过滤相结合的混合策略，借助用户-酒款交互矩阵、SVD矩阵分解和余弦相似度计算完成个性化推荐。资源以单个docx文档呈现，大小约128KB，内含完整代码示例、系统架构、目录结构、设计思路与关键模块详解，从数据生成和清洗、特征工程到模型训练、评估与可视化，再到前后端交互，形成可落地的闭环智能推荐应用。已有66人学习下载，适合边阅读边动手搭建，可作为从理论到实践的完整项目范例，并依据实际业务需求进行二次开发与调优。

1. 先看清这套酒庄推荐系统：一个能跑的完整闭环

做电商或内容推荐的人都有这种体验：算法文章读了一堆，真到自己搭项目时，卡在数据清洗和前后端对接上。这套酒庄推荐系统是个例外，它把 Python 数据分析、MySQL 存储、FastAPI 接口和 Tkinter 桌面端串成了一个能实际运行的完整闭环。拿到手先用自带的数据生成脚本造一批模拟订单和行为日志，从构建交互矩阵、SVD 协同过滤、内容过滤一路推到冷启动规则，最后在 GUI 界面看到每个用户的个性化酒款列表。适合有一定 Python 基础、想完整走一遍推荐系统落地流程的开发者，也适合准备把推荐能力迁移到商城或内容平台的从业者。整套项目按软件工程组织：需求分析、MySQL 表结构、API 规范、后端服务、前端界面、部署监控都在，推荐部分用协同过滤加内容过滤的混合策略，交互多的用户走 SVD，新用户由热门和新品规则兜底，冷启动路由是实际业务里最影响体验的一环。

2. 数据处理链路：从多源脏数据到用户–酒款交互矩阵

2.1 数据源盘点与字段标准化

酒庄的数据来源很杂：电商平台导出的订单、线下 POS 销售小票、会员系统的等级与积分、品鉴会报名记录、第三方酒评网站的评分描述。各系统字段命名不一致，日期格式各异，ID 类型也不统一——有的用自增数字，有的用带前缀的字符串。项目里第一步不是急着训练模型，而是先把这些来源按实体拆开：用户、酒款、行为日志、订单，分别做字段标准化。

常见的做法是写一个归一化函数，把列名统一成小写下划线风格，日期强制转成 datetime 类型，业务主键统一转成字符串。这一步很细碎，但漏了它，后面 groupby 和 join 全都会出问题。

import pandas as pd import numpy as np def normalize_frames(wine_df, user_df, log_df): # 酒款表字段统一 wine_df = wine_df.rename(columns={ "酒款ID": "wine_id", "酒款名称": "wine_name", "葡萄品种": "grape_variety", "产区": "region", "年份": "vintage", "酒精度": "abv", "价格": "price", "评分": "rating"}) # 行为日志时间字段统一，解析失败置为 NaT log_df["log_time"] = pd.to_datetime(log_df["log_time"], errors="coerce") # 所有主键统一转字符串，避免数字ID和字符串ID拼接时错位 for df in (wine_df, user_df, log_df): for col in ("wine_id", "user_id"): if col in df.columns: df[col] = df[col].astype(str) return wine_df, user_df, log_df

这里有两个容易被忽略的细节。一是errors="coerce"，日期格式乱的数据会被置成 NaT，后续统一处理，而不是让程序直接抛异常中断；二是主键全部转字符串，否则 CSV 里“1001”和“01001”会被 pandas 当成两个不同的 ID，多源合并时会产生重复用户。

2.2 主键对齐、缺失值与异常值处理

多源数据合并前要建映射表。比如线下会员系统里用户 ID 是M00123，电商平台里是123，需要先写一张映射关系，把两边 ID 拉到同一套主键下。项目文档里专门提到这一步，实际操作中也最容易翻车：跳过对齐直接 concat，会出现同一用户被切成两个人，推荐结果被稀释。

缺失值和异常值按字段类型分别处理。价格、评分这类数值字段通常用中位数填充，受极端值影响小；葡萄品种、产区这类类别字段用众数填充；时间字段缺失的直接按默认时间补齐或者在统计时排除。异常值用箱线图规则和 3 倍标准差规则双重检测，把异常价格、超出合理范围的酒精度修正掉，而不是直接删行。

def clean_wine(wine_df): # 价格缺失用中位数，品种缺失用众数 wine_df["price"] = wine_df["price"].fillna(wine_df["price"].median()) wine_df["grape_variety"] = wine_df["grape_variety"].fillna( wine_df["grape_variety"].mode()[0]) # 3倍标准差检测价格异常，异常值回填中位数 mean, std = wine_df["price"].mean(), wine_df["price"].std() mask = np.abs(wine_df["price"] - mean) > 3 * std wine_df.loc[mask, "price"] = wine_df["price"].median() # 评分范围校验，超出0-100区间的按边界截断 wine_df["rating"] = wine_df["rating"].clip(0, 100) return wine_df

填充策略不是随便选的，价格字段偏态明显，用中位数比均值稳；评分字段有业务边界，clip 截断比删除更合理，保留样本量。整套清洗管道可以封装成函数，在数据更新后重复执行，保证每次建模前拿到的都是同一套规则处理过的数据。

2.3 隐式评分构建与用户–酒款交互矩阵

酒庄业务里几乎没有显式评分数据，用户很少给酒打星。项目把浏览、收藏、加购物车、下单这些行为日志转成隐式评分，用不同权重表达行为强度：浏览是浅层兴趣，加购和下单才是强偏好信号。权重不是拍脑袋定的，通常先按业务经验设初值，再通过推荐效果反馈调。

# 行为类型到隐式评分的映射 weight_map = {"view": 1, "favorite": 2, "cart": 3, "order": 5} log_df["score"] = log_df["behavior_type"].map(weight_map).fillna(1) log_df["score"] = log_df["score"] * log_df["behavior_num"].fillna(1) # 同一用户对同一酒款多次行为做加权求和 user_item = (log_df.groupby(["user_id", "wine_id"])["score"] .sum().reset_index()) # 透视成交互矩阵，行=用户，列=酒款，无行为填0 matrix = user_item.pivot_table( index="user_id", columns="wine_id", values="score").fillna(0) # 稀疏度检查：非零元素占比 sparsity = 1 - (matrix > 0).sum().sum() / matrix.size print(f"交互矩阵稀疏度: {sparsity:.2%}")

交互矩阵是后面所有推荐算法的基础输入。透视后行是用户、列是酒款，每个格子的值代表用户对酒款的隐式偏好强度。fillna(0) 在这里不是随便补零，而是明确表达“没有行为记录不代表讨厌，只是未知”，这也是矩阵分解模型的前提假设之一。跑完透视后先打印稀疏度，如果超过 99%，后续 SVD 的效果大概率有限，需要回去检查行为日志的覆盖度。

3. 混合推荐模型：SVD 协同过滤、内容过滤与冷启动兜底

3.1 为什么是混合推荐：稀疏矩阵下的选型逻辑

酒庄场景有两个天然难点。一是用户购买频次低，大部分人一年在酒庄下单十几次，行为日志稀疏；二是酒款长尾严重，几百款酒里可能只有二三十款有稳定销量，剩下的分布在长长的尾部。这种情况下单独用基于邻域的协同过滤，很多用户和酒款找不到足够近邻，推荐结果不稳定；单独用内容过滤，又只会推荐用户买过的同品类，跨品类发现能力弱。

项目采用混合策略：交互数据丰富的用户走 SVD 矩阵分解，通过隐因子弥补稀疏性；交互少的用户走内容过滤，基于酒款属性找相似；新用户和新酒款由规则兜底。三套结果按用户的历史行为量动态路由，而不是对所有人跑同一套模型。这个路由判断，配合后面 API 接口里的行为量阈值，是整个推荐服务的核心骨架。

3.2 协同过滤：SVD 矩阵分解的用户–酒款隐因子

矩阵分解的思路是把交互矩阵拆成两个低维矩阵的乘积，一个代表用户隐因子，一个代表酒款隐因子。对酒庄场景，隐因子可以理解为某种抽象口味维度：旧世界 vs 新世界、浓郁 vs 轻盈、节日送礼 vs 日常自饮。模型本身不会给这些维度命名，但训练出的向量能表达这种差异，这也是后面做相似酒款推荐的基础。

surprise 库的 SVD 适合做带评分的矩阵分解，但这里交互矩阵是隐式评分，直接用 scikit-learn 的 TruncatedSVD 更顺手，输出就是 numpy 矩阵，方便嵌入 FastAPI 做在线推理。

from sklearn.decomposition import TruncatedSVD # 隐因子数量：用户量和酒款量在500级别时，32~64维足够 svd = TruncatedSVD(n_components=32, random_state=42) U = svd.fit_transform(matrix) # 用户困因子，shape=(n_users, 32) Vt = svd.components_ # 酒款隐因子，shape=(32, n_wines) pred_matrix = np.dot(U, Vt) # 还原的完整预测矩阵

n_components不是越大越好，酒款总量可能只有几百，取 64 以上容易把噪声也学进去；random_state固定是为了让每次训练结果可复现，否则调参时没法对比效果。推理时对某个用户，只需要取 U 里对应行和 Vt 做矩阵乘法，不用重新训练，单次查询开销很小。

3.3 内容过滤：酒款特征向量与余弦相似度

内容过滤不依赖用户历史，而是基于酒款自身的可描述属性：葡萄品种、产区、年份、酒精度、价格区间、评分。处理这类特征时，类别字段用 One-Hot 编码，数值字段做 MinMax 归一化，然后拼接成统一维度的特征向量，用余弦相似度衡量酒款间的相似程度。

from sklearn.preprocessing import OneHotEncoder, MinMaxScaler from sklearn.metrics.pairwise import cosine_similarity # 类别特征：品种、产区做 One-Hot cat_feats = OneHotEncoder(sparse_output=False).fit_transform( wine_df[["grape_variety", "region"]]) # 数值特征：价格、酒精度、评分统一归一化到[0,1] num_feats = MinMaxScaler().fit_transform( wine_df[["price", "abv", "rating"]]) # 拼接成完整特征向量 feature_matrix = np.hstack([cat_feats, num_feats]) sim_matrix = cosine_similarity(feature_matrix) def content_topk(wine_id: str, k: int = 10): idx = wine_id_to_idx[wine_id] sim_scores = sim_matrix[idx] top_idx = np.argsort(-sim_scores)[1:k + 1] return [idx_to_wine_id[i] for i in top_idx]

One-Hot 编码后品种和产区维度的稀疏性较高，余弦相似度对这种特征分布比欧氏距离更稳定。实际使用时会发现，纯内容过滤的结果容易局限在同一品种或同一产区，所以项目里把它作为用户行为较少时的初筛，不单独作为最终推荐。argsort(-sim_scores)取的是从大到小排序，跳过第一个是因为它一定是酒款自身，相似度为 1。

3.4 冷启动规则：热门榜、新品榜与交互阈值路由

新用户没有行为记录，新酒款没有交互数据，这两个场景必须靠规则。项目里的冷启动策略很务实，按用户行为量分档处理，文档中也给了这套规则的路由逻辑。

阈值 5 不是固定值，需要根据交互矩阵的平均行密度调整。如果整体行为日志很少，可以降到 3；如果用户行为普遍密集，可以提到 8。规则层在 FastAPI 接口里是最后一道兜底，确保任何用户访问推荐接口时都能拿到非空列表。

def cold_start_recommend(user_id: str, top_k: int = 10): behavior_count = count_user_behaviors(user_id) if behavior_count == 0: # 新用户：注册时选的偏好品种 + 全站热门 pref = get_user_pref(user_id) return hot_wines(limit=top_k, seed=pref) liked = recent_clicked_wines(user_id) candidates = [] for w in liked[:3]: # 只取最近3个感兴趣酒款做扩散 candidates += content_topk(w, k=20) return dedupe(candidates)[:top_k]

这套规则的核心是“先用内容过滤扩散，再用热门兜底”。只取最近 3 个酒款做扩展是防止用户短时间误点太多，导致推荐列表被噪声带偏；去重后截断到 top_k 是为了保证列表长度一致。实际项目中这三层逻辑会以模块形式独立存在，便于在路由判断处灵活切换。

4. 工程落地：MySQL 表设计、API 接口与 FastAPI 服务

4.1 MySQL 核心表设计：用户、酒款、行为、订单与推荐缓存

项目的数据库按业务对象拆成五张核心表：用户会员表、酒款商品表、行为日志表、订单主表和明细表、推荐结果缓存与模型版本表。早期示例代码里有 Flask 写的推荐服务接口，适合快速演示在线推理；完整工程后端主体换成了 FastAPI，自带请求参数校验和异步支持，部署更省事。数据库结构两边共用，不改动。

CREATE TABLE users ( user_id BIGINT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(50) NOT NULL UNIQUE, password_hash VARCHAR(255) NOT NULL, member_level TINYINT DEFAULT 0 COMMENT '0普通 1银卡 2金卡', preferred_grapes VARCHAR(100) COMMENT '注册时选的偏好品种', created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE wines ( wine_id BIGINT PRIMARY KEY AUTO_INCREMENT, wine_name VARCHAR(100) NOT NULL, grape_variety VARCHAR(50), region VARCHAR(50), vintage SMALLINT COMMENT '年份', abv DECIMAL(3,1) COMMENT '酒精度', price DECIMAL(10,2), rating DECIMAL(4,2) COMMENT '酒评家评分', on_sale TINYINT DEFAULT 1, created_at DATETIME ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

两张核心表写出来，能看出几个刻意设计。一是字符集统一 utf8mb4，酒款名称和产区描述里经常有中文特殊字符，utf8mb4 才能完整存下；二是会员等级、偏好品种直接冗余在用户表里，冷启动推荐时不用再去 JOIN 行为日志；三是价格用 DECIMAL 而不是 FLOAT，避免浮点精度导致金额对不上。行为日志表和订单表的要点在于索引：行为日志按user_id + log_time建联合索引，订单明细按order_id + wine_id建联合索引，因为查询推荐和统计销量都走这两个路径。

4.2 API 接口规范：路径、入参与出参

项目文档里定义了一套完整的 REST API，前后端按这套规范联调。核心接口集中在用户、推荐、酒款、订单和报表五类，下面这张表是实际开发中最常用的部分。

个性化推荐接口专门设计了一个version参数，用于指定用哪个模型版本出结果。这个参数在平时没有存在感，但模型迭代时非常重要——线上跑着的服务在切换版本时，可以用它做 A/B 对照或快速回滚。推荐理由字段是项目里的可解释性设计，接口返回每个酒款时要带上“因为您最近浏览了赤霞珠”这类说明，前端直接展示。

4.3 FastAPI 后端：模型加载、推荐路由与结果缓存

后端服务启动时一次性加载训练好的模型文件，避免每个请求都重新读磁盘。推荐接口先查用户行为量，再决定走 SVD 还是冷启动规则，最后把结果写入推荐结果缓存表，带模型版本号。

import joblib import numpy as np from fastapi import FastAPI, HTTPException app = FastAPI(title="wine-recommend-api") # 启动时加载模型，SVD训练一次后序列化落盘 model = joblib.load("../models/svd_v2.1.pkl") U, Vt, wine_ids, user_index = (model["U"], model["Vt"], model["wine_ids"], model["user_index"]) @app.get("/api/recommend/{user_id}") def recommend(user_id: str, top_k: int = 10): user_id = str(user_id) behavior_count = count_user_behaviors(user_id) if behavior_count < 5: items = cold_start_recommend(user_id, top_k) # 规则兜底 else: items = svd_predict(user_id, top_k) # 隐因子预测 if not items: raise HTTPException(status_code=404, detail="no recommend result") cache_recommend(user_id, items, model_version="svd_v2.1") return {"user_id": user_id, "items": items, "model_version": "svd_v2.1"} def svd_predict(user_id: str, top_k: int): if user_id not in user_index: return [] u_vec = U[user_index[user_id]] # 用户隐因子向量 scores = u_vec @ Vt # 与全部酒款内积打分 top_idx = np.argsort(-scores)[:top_k] return [wine_ids[i] for i in top_idx]

启动时加载模型是关键动作，如果模型文件损坏或路径错了，服务应该在启动阶段报错，而不是等到用户请求时才失败。u_vec @ Vt是对整个酒款集合的一次矩阵乘法，比循环逐个预测快一个量级，在酒款数量几千以内的场景单次请求在毫秒级。cache_recommend写入缓存表时会带上模型版本号，这就是后面做推荐回归对比的数据来源。

5. 系统联调与避坑：Tkinter 前端 GUI 和四个高频问题

5.1 前端整体结构与窗口管理

前端用 Tkinter 实现了五个窗口：登录窗口、主界面、酒款搜索、订单创建、历史订单查看。窗口之间通过主界面的菜单栏整合，登录成功后跳转主界面，主界面里再打开子窗口。项目里把每个窗口封装成独立类，类内部只负责自己的 UI 布局和请求逻辑，窗口间通过传 token 和 user_id 保持状态，这与后端按模块划分是对应的。

import tkinter as tk from tkinter import ttk, messagebox import requests API_BASE = "http://127.0.0.1:8000" class LoginWindow: def __init__(self): self.root = tk.Tk() self.root.title("酒庄推荐系统 - 登录") self.token = None self.user_id = None tk.Label(self.root, text="用户名").grid(row=0, column=0, padx=5, pady=5) self.username = tk.Entry(self.root) self.username.grid(row=0, column=1, padx=5, pady=5) tk.Label(self.root, text="密码").grid(row=1, column=0, padx=5, pady=5) self.password = tk.Entry(self.root, show="*") self.password.grid(row=1, column=1, padx=5, pady=5) tk.Button(self.root, text="登录", command=self.do_login).grid(row=2, columnspan=2)

登录窗口的关键在do_login方法里，请求要放在子线程中执行，不能直接在按钮回调里做同步 requests，否则后端慢一点，整个 UI 就卡死。项目里的做法是先禁掉登录按钮，再开线程请求，拿到结果后通过回调回到主线程更新界面。

5.2 主界面推荐展示与子窗口交互

主界面用 Treeview 展示推荐结果，列分别显示酒款 ID、名称、价格和推荐理由。推荐理由来自后端接口的reason字段，前端直接原样展示，这是项目里可解释性设计在 UI 层落地。搜索窗口和订单窗口通过主界面传入的 token 调接口，下单成功后刷新推荐列表，形成业务闭环。

class MainWindow(tk.Tk): def __init__(self, token, user_id): super().__init__() self.title("酒庄数据分析推荐系统") self.token = token self.user_id = user_id cols = ("wine_id", "wine_name", "price", "reason") self.tree = ttk.Treeview(self, columns=cols, show="headings") for c in cols: self.tree.heading(c, text=c) self.tree.pack(fill="both", expand=True) tk.Button(self, text="刷新推荐", command=self.load_recommend).pack() def load_recommend(self): headers = {"Authorization": f"Bearer {self.token}"} url = f"{API_BASE}/api/recommend/{self.user_id}" resp = requests.get(url, headers=headers, timeout=5) if resp.status_code != 200: messagebox.showerror("错误", resp.text) return self.tree.delete(*self.tree.get_children()) for item in resp.json()["items"]: self.tree.insert("", "end", values=( item["wine_id"], item["wine_name"], item["price"], item["reason"]))

请求加了timeout=5是联调阶段最容易漏的配置，没有超时的话，后端线程池一满，界面会永远转圈。刷新时先清空 Treeview 再插入，避免旧数据和新数据混在一起误导用户。推荐接口返回的items里每一项都携带reason字段，这个字段在内网负载低时生成很快，但如果数据库查询慢，会成为接口的瓶颈。

5.3 实战避坑：GUI 联调的四个高频问题

这套项目从零跑通，最容易在联调环节翻车，下面按踩坑频率排序。

• 现象：点击“刷新推荐”后窗口无响应，转圈几分钟后才弹结果，甚至直接假死。 原因：Tkinter 主线程里执行了同步 requests 请求，GUI 事件循环被阻塞，后端只要超过 1 秒就明显卡顿。 解决：所有网络请求扔进ThreadPoolExecutor，请求完成后再回到主线程刷新界面，同时请求加 timeout。

• 现象：新注册用户进入推荐页，接口返回 404 或空列表。 原因：用户没有任何行为日志，SVD 的user_index里没有这个 ID，直接访问模型拿不到嵌入，冷启动规则没接上。 解决：推荐接口先查行为量，为 0 的走热门兜底，行为量少的走内容过滤，保证任何用户都能拿到非空结果。

• 现象：数据库里中文酒款名和产区显示乱码，个别像“Château”这类带特殊字符的字符串入库报错。 原因：表结构没指定 utf8mb4，或 MySQL 连接串没加 charset 参数，默认 latin1 存不下中文和扩展字符。 解决：建表统一DEFAULT CHARSET=utf8mb4，连接串加上?charset=utf8mb4，后端代码里不手动转码。

• 现象：服务启动成功，但第一个推荐请求就报“Not fitted”或模型属性找不到。 原因：模型加载代码写在请求函数内部，或者模型文件路径写死但部署环境里不存在，异常被吞掉后请求时才发现。 解决：模型文件在应用启动阶段加载并做有效性校验，文件缺失直接让服务启动失败，而不是带病运行。

这几条全是实际跑这套系统时真实会碰到的问题，每一条都能通过日志定位到具体环节。排查顺序建议从后往前：先看接口返回状态码，再看后端日志有没有报错关键字，最后查数据库表数据是否写入成功。

6. 效果验证与迭代技巧：离线指标、TopK 回归对比与版本管理

6.1 离线评估指标怎么算

推荐模型上线前先算离线指标。项目里用的是三维度：评分预测准度用 RMSE，列表质量用 TopK 的 precision 和 recall，系统多样性看推荐列表的类别覆盖。RMSE 低只代表预测值贴近真实值，不代表用户会点；酒庄场景更该关注 precision@10，因为用户每天看到的推荐位就那么多，前十个准不准直接影响点击率和转化率。

6.2 用推荐缓存表做模型版本回归

项目里的推荐结果缓存表带了 model_version 字段，这个字段除了排查问题，还能做模型迭代时的回归检查。每次训练新模型，先跑一批老用户的推荐结果，和线上版本做 TopK 重合度对比，重合度太低说明行为变化大，需要人工确认是数据变了还是模型坏了。用 pandas 直接读缓存表就能算：

import pandas as pd from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://root:password@localhost/wine_db?charset=utf8mb4") df_old = pd.read_sql( "SELECT user_id, wine_id FROM recommend_cache WHERE model_version='v2.0'", engine) df_new = pd.read_sql( "SELECT user_id, wine_id FROM recommend_cache WHERE model_version='v2.1'", engine) def jaccard_of_sets(set_a, set_b): if not set_a and not set_b: return 1.0 return len(set_a & set_b) / len(set_a | set_b) # 对比单个用户的 Top10 重合度 old_set = set(df_old[df_old["user_id"] == 10001]["wine_id"]) new_set = set(df_new[df_new["user_id"] == 10001]["wine_id"]) print(f"Top10 重合度: {jaccard_of_sets(old_set, new_set):.2%}")

重合度在 0.7 以上意味着新版本基本保留了老用户的口味结构，可以放心灰度；低于 0.4 就要警惕，可能是 SVD 隐因子数量改得太激进，也可能是训练数据分布变了。从这个技巧延伸出去，还可以对user_id分组批量计算平均重合度，作为一次模型更新的整体评估。

从那以后，我每次改完模型参数或更新酒款数据，都会强制走一遍同样的流程：先离线算 RMSE 和 TopK 重合度，再查一遍推荐缓存表里有没有异常版本，最后才允许自己动接口配置。这个习惯是从一次线上推荐全变、回滚了三次才定位到原因的经验里换来的，希望帮到你。

本文还有配套的精品资源，点击获取