1. 项目背景与核心价值

理财推荐系统是金融科技领域的热门应用方向。传统金融机构在向客户推荐理财产品时，往往面临两个痛点：一是人工推荐效率低下，难以覆盖海量客户；二是标准化推荐缺乏个性化，难以匹配客户真实需求。基于协同过滤算法的推荐系统能有效解决这些问题。

我在某金融科技公司实习期间，曾参与过银行理财推荐系统的升级项目。当时行内使用的还是基于规则引擎的推荐逻辑，转化率长期徘徊在3%左右。改用协同过滤算法后，首月转化率就提升到了8.2%，这让我深刻认识到算法推荐在金融领域的价值。

这个Python实现的协同过滤理财推荐系统，具有以下典型应用场景：

• 银行APP的"猜你喜欢"板块
• 理财顾问的智能辅助工具
• 第三方理财平台的个性化首页
• 金融教育平台的学练结合推荐

2. 系统架构设计

2.1 整体技术栈

系统采用经典的三层架构：

表示层：Bootstrap3 + Django模板 业务层：Django框架 + 协同过滤算法 数据层：MySQL + Redis缓存

选择这套技术栈主要基于以下考虑：

1. Django自带Admin后台，非常适合快速开发管理系统
2. Bootstrap3的响应式布局能适配移动端和PC端
3. MySQL作为成熟的关系型数据库，完全能满足理财产品的结构化存储需求
4. Redis缓存用户行为数据，大幅提升推荐实时性

2.2 数据库设计

核心表结构设计如下：

用户表(users)

CREATE TABLE `users` ( `user_id` int(11) NOT NULL AUTO_INCREMENT, `username` varchar(50) NOT NULL, `risk_level` enum('保守型','稳健型','平衡型','成长型','进取型') NOT NULL, `register_time` datetime NOT NULL, PRIMARY KEY (`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

理财产品表(products)

CREATE TABLE `products` ( `product_id` int(11) NOT NULL AUTO_INCREMENT, `product_name` varchar(100) NOT NULL, `product_type` enum('货币型','债券型','混合型','股票型','QDII') NOT NULL, `expected_return` decimal(5,2) NOT NULL, `risk_level` enum('R1','R2','R3','R4','R5') NOT NULL, `min_amount` decimal(12,2) NOT NULL, PRIMARY KEY (`product_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

用户行为表(user_behavior)

CREATE TABLE `user_behavior` ( `id` int(11) NOT NULL AUTO_INCREMENT, `user_id` int(11) NOT NULL, `product_id` int(11) NOT NULL, `behavior_type` enum('浏览','收藏','购买','赎回') NOT NULL, `behavior_time` datetime NOT NULL, `weight` decimal(3,2) NOT NULL COMMENT '行为权重', PRIMARY KEY (`id`), KEY `idx_user_product` (`user_id`,`product_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

注意：行为权重是协同过滤的关键参数，需要根据业务经验设定。通常购买=1.0，收藏=0.6，浏览=0.3。这个参数会显著影响推荐效果，需要在实际运营中持续优化。

3. 协同过滤算法实现

3.1 用户-产品评分矩阵构建

核心是构建用户对产品的评分矩阵。我们采用加权行为数据作为评分依据：

def build_rating_matrix(): # 从数据库加载原始行为数据 behaviors = UserBehavior.objects.all().values( 'user_id', 'product_id', 'weight') # 转换为字典结构 {user_id: {product_id: score}} rating_dict = defaultdict(dict) for b in behaviors: user_id = b['user_id'] product_id = b['product_id'] if product_id in rating_dict[user_id]: rating_dict[user_id][product_id] += b['weight'] else: rating_dict[user_id][product_id] = b['weight'] # 归一化处理（0-5分制） for user_id in rating_dict: max_score = max(rating_dict[user_id].values()) for product_id in rating_dict[user_id]: rating_dict[user_id][product_id] = round( rating_dict[user_id][product_id]/max_score*5, 2) return rating_dict

3.2 相似度计算

采用改进的余弦相似度计算用户相似度，加入风险偏好修正因子：

def cosine_sim(user1, user2, risk_weight=0.3): """ 计算两个用户的相似度 :param user1: 用户1的评分字典 {product_id: score} :param user2: 用户2的评分字典 {product_id: score} :param risk_weight: 风险偏好相似度的权重 :return: 0-1之间的相似度值 """ # 获取共同评价过的产品 common_products = set(user1.keys()) & set(user2.keys()) if not common_products: return 0 # 计算评分余弦相似度 dot_product = sum(user1[p] * user2[p] for p in common_products) norm1 = math.sqrt(sum(user1[p]**2 for p in common_products)) norm2 = math.sqrt(sum(user2[p]**2 for p in common_products)) rating_sim = dot_product / (norm1 * norm2) # 获取用户风险偏好（从数据库查询） risk1 = User.objects.get(id=user1.id).risk_level risk2 = User.objects.get(id=user2.id).risk_level risk_levels = ['保守型', '稳健型', '平衡型', '成长型', '进取型'] risk_sim = 1 - abs(risk_levels.index(risk1) - risk_levels.index(risk2)) / 4 # 加权综合相似度 return (1-risk_weight) * rating_sim + risk_weight * risk_sim

3.3 推荐生成

基于用户的协同过滤推荐核心逻辑：

def recommend_products(target_user_id, rating_dict, n=10): # 计算目标用户与其他用户的相似度 similarities = [] for user_id in rating_dict: if user_id == target_user_id: continue sim = cosine_sim(rating_dict[target_user_id], rating_dict[user_id]) similarities.append((user_id, sim)) # 按相似度降序排序 similarities.sort(key=lambda x: x[1], reverse=True) # 取Top50相似用户 top_users = [user_id for user_id, sim in similarities[:50]] # 收集相似用户喜欢但目标用户未接触的产品 recommendations = defaultdict(float) target_products = set(rating_dict[target_user_id].keys()) for user_id in top_users: for product_id in rating_dict[user_id]: if product_id not in target_products: # 相似度加权评分 recommendations[product_id] += ( rating_dict[user_id][product_id] * next(sim for uid, sim in similarities if uid == user_id) ) # 按加权评分降序排序 sorted_recommends = sorted(recommendations.items(), key=lambda x: x[1], reverse=True) # 返回TopN推荐产品ID return [product_id for product_id, score in sorted_recommends[:n]]

4. 系统实现关键点

4.1 冷启动问题解决方案

理财推荐系统面临严重的冷启动问题：

• 新用户没有行为数据
• 新产品没有被购买记录

我们采用混合推荐策略解决：

1. 基于内容的过滤：新产品根据其类型、风险等级匹配相似产品
2. 规则引擎兜底：新用户根据注册时填写的风险测评结果推荐

def hybrid_recommend(user_id, rating_dict): # 检查用户是否有足够行为数据 if user_id not in rating_dict or len(rating_dict[user_id]) < 5: # 冷启动情况 user = User.objects.get(id=user_id) # 方法1：基于风险等级的规则推荐 risk_map = { '保守型': ['货币型', '债券型'], '稳健型': ['债券型', '混合型'], '平衡型': ['混合型'], '成长型': ['混合型', '股票型'], '进取型': ['股票型', 'QDII'] } products = Product.objects.filter( product_type__in=risk_map[user.risk_level] ).order_by('-expected_return')[:10] return [p.product_id for p in products] else: # 正常协同过滤推荐 return recommend_products(user_id, rating_dict)

4.2 实时推荐优化

传统协同过滤算法通常是离线计算的，我们通过以下方式实现准实时推荐：

1. 用户行为数据写入MySQL的同时写入Redis
2. 每小时全量更新一次评分矩阵
3. 当用户访问推荐接口时，先检查Redis中是否有最新行为
4. 如果有新行为，则实时更新内存中的用户评分向量

def get_realtime_rating(user_id): # 从内存获取基础评分数据 user_ratings = copy.deepcopy(rating_dict.get(user_id, {})) # 检查Redis中的最新行为 redis_key = f"recent_behavior:{user_id}" recent_behaviors = redis_client.lrange(redis_key, 0, -1) # 更新评分 for behavior in recent_behaviors: product_id, weight = behavior.decode().split(':') product_id = int(product_id) weight = float(weight) if product_id in user_ratings: user_ratings[product_id] = min(5, user_ratings[product_id] + weight) else: user_ratings[product_id] = min(5, weight * 5) return user_ratings

4.3 多样性保障机制

协同过滤容易导致推荐结果同质化。我们引入三大机制保障多样性：

1. 类型多样性：确保推荐列表中包含至少3种不同类型产品
2. 风险分散：推荐产品的风险等级不超过用户风险等级的±1级
3. 新颖性注入：每天随机选择5%的流量尝试推荐上市不足30天的新品

def diversify_recommendations(product_ids, user_id): user = User.objects.get(id=user_id) products = Product.objects.filter(product_id__in=product_ids) # 按类型分组 type_groups = defaultdict(list) for p in products: type_groups[p.product_type].append(p.product_id) # 确保至少3种类型 if len(type_groups) < 3: needed_types = set(['货币型', '债券型', '混合型']) - set(type_groups.keys()) for t in needed_types: extra = Product.objects.filter( product_type=t, risk_level__lte=user.risk_level ).order_by('-expected_return')[:1] if extra: product_ids.append(extra[0].product_id) # 风险等级过滤 risk_levels = ['R1', 'R2', 'R3', 'R4', 'R5'] user_risk_index = ['保守型', '稳健型', '平衡型', '成长型', '进取型'].index(user.risk_level) allowed_risks = risk_levels[max(0, user_risk_index-1):user_risk_index+2] product_ids = [pid for pid in product_ids if Product.objects.get(product_id=pid).risk_level in allowed_risks] return product_ids[:10] # 最终返回前10个

5. 系统部署与性能优化

5.1 部署架构

生产环境推荐使用以下部署方案：

前端服务器：Nginx + uWSGI (2核4G) 应用服务器：Django + Gunicorn (4核8G，建议2-4个worker) 数据库服务器：MySQL主从 (8核16G，SSD磁盘) 缓存服务器：Redis哨兵模式 (4核8G)

5.2 性能优化技巧

1. 评分矩阵缓存：将用户-产品评分矩阵缓存在Redis中，每小时更新一次
2. 相似度预计算：每天凌晨计算活跃用户之间的相似度并缓存
3. 异步日志：用户行为日志采用异步写入方式，避免阻塞主流程
4. 数据库索引优化：确保user_behavior表有(user_id, product_id)联合索引
5. 连接池配置：数据库和Redis都使用连接池，避免频繁创建连接

# Django的数据库连接池配置示例 DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'NAME': 'finance_recommend', 'USER': 'recommend_user', 'PASSWORD': 'securepassword', 'HOST': 'mysql-master', 'PORT': '3306', 'OPTIONS': { 'pool_size': 20, 'max_overflow': 10, 'pool_timeout': 30, } } }

5.3 压力测试结果

使用Locust进行压力测试，单服务器配置(4核8G)下的性能表现：

实际部署建议：当并发预计超过200时，应该考虑水平扩展应用服务器。数据库层面，当用户量超过50万时，需要考虑分库分表策略。

6. 效果评估与调优

6.1 评估指标体系

理财推荐系统的效果评估需要综合多个指标：

1. 点击率(CTR)：推荐产品被点击的比例
2. 转化率(Conversion Rate)：推荐产品最终被购买的比例
3. 多样性(Diversity)：推荐列表中不同类型产品的分布
4. 新颖性(Novelty)：推荐产品中有多少是用户从未接触过的
5. 覆盖率(Coverage)：系统能推荐的产品占全部产品的比例

6.2 A/B测试方案

我们设计了以下A/B测试策略：

对照组：原有规则引擎推荐策略

• 根据用户风险等级推荐同类型产品
• 按预期收益率从高到低排序

实验组：协同过滤推荐策略

• 基于用户行为数据的协同过滤
• 加入多样性保障机制

测试周期为2周，关键结果对比如下：

6.3 常见问题与调优

问题1：热门产品过度推荐

• 现象：少数热销产品占据大部分推荐位
• 解决方案：引入流行度惩罚因子，降低热门产品的推荐权重

def apply_popularity_penalty(product_scores): # 获取产品流行度（购买次数） popularities = Product.objects.annotate( popularity=Count('userbehavior') ).values('product_id', 'popularity') pop_dict = {p['product_id']: p['popularity'] for p in popularities} max_pop = max(pop_dict.values()) if pop_dict else 1 # 应用惩罚因子 penalized_scores = {} for pid, score in product_scores.items(): penalty = 0.8 + 0.2 * (1 - pop_dict.get(pid, 0)/max_pop) penalized_scores[pid] = score * penalty return penalized_scores

问题2：风险错配

• 现象：激进型用户偶尔收到保守型产品推荐
• 解决方案：在相似度计算中加大风险偏好的权重，并添加后置过滤

问题3：季节效应

• 现象：年末货币基金推荐效果突然变差
• 解决方案：引入时间衰减因子，近期的行为权重更高

def apply_time_decay(user_ratings): # 获取用户最近行为时间 latest_time = UserBehavior.objects.filter( user_id=user_id ).latest('behavior_time').behavior_time # 计算时间衰减 decayed_ratings = {} for pid, score in user_ratings.items(): behavior_time = UserBehavior.objects.filter( user_id=user_id, product_id=pid ).latest('behavior_time').behavior_time days_diff = (latest_time - behavior_time).days decay_factor = 0.9 ** days_diff # 每天衰减10% decayed_ratings[pid] = score * decay_factor return decayed_ratings

7. 项目扩展方向

7.1 加入深度学习模型

传统协同过滤可以升级为神经协同过滤(NCF)：

1. 使用神经网络学习用户和产品的嵌入表示
2. 引入注意力机制捕捉不同行为的重要性差异
3. 结合元学习处理冷启动问题

# 简易NCF模型示例 from tensorflow.keras.layers import Input, Embedding, Flatten, Concatenate, Dense def build_ncf_model(num_users, num_products, embedding_size=64): # 输入层 user_input = Input(shape=(1,)) product_input = Input(shape=(1,)) # 嵌入层 user_embedding = Embedding(num_users, embedding_size)(user_input) user_embedding = Flatten()(user_embedding) product_embedding = Embedding(num_products, embedding_size)(product_input) product_embedding = Flatten()(product_embedding) # 交互层 concat = Concatenate()([user_embedding, product_embedding]) # 全连接层 dense1 = Dense(128, activation='relu')(concat) dense2 = Dense(64, activation='relu')(dense1) output = Dense(1, activation='sigmoid')(dense2) # 构建模型 model = Model(inputs=[user_input, product_input], outputs=output) model.compile(optimizer='adam', loss='binary_crossentropy') return model

7.2 多目标优化

理财推荐不应只关注转化率，还需要考虑：

1. 用户资产配置的合理性
2. 产品之间的风险对冲
3. 用户长期价值最大化

可以设计多目标优化框架：

def multi_objective_optimization(product_ids, user_id): objectives = { 'conversion': predict_conversion_prob(user_id, product_ids), 'diversity': calculate_diversity(product_ids), 'risk_balance': assess_risk_balance(user_id, product_ids), 'long_term_value': estimate_long_term_value(user_id, product_ids) } # 使用加权求和法 weights = { 'conversion': 0.4, 'diversity': 0.2, 'risk_balance': 0.3, 'long_term_value': 0.1 } scores = [] for pid in product_ids: score = sum(objectives[obj][pid] * weights[obj] for obj in objectives) scores.append((pid, score)) return sorted(scores, key=lambda x: x[1], reverse=True)

7.3 可视化分析平台

构建推荐效果可视化看板，监控：

1. 实时推荐流量分布
2. 转化漏斗分析
3. 用户分群推荐效果
4. 产品推荐热度图

使用Django+ECharts实现示例：

# views.py def dashboard(request): # 获取最近7天数据 stats = RecommendationStats.objects.filter( date__gte=timezone.now()-timedelta(days=7) ).values('date').annotate( ctr=Avg('click_rate'), conversion=Avg('conversion_rate') ).order_by('date') dates = [s['date'].strftime('%m-%d') for s in stats] ctr_data = [float(s['ctr']) for s in stats] conversion_data = [float(s['conversion']) for s in stats] return render(request, 'dashboard.html', { 'dates': json.dumps(dates), 'ctr_data': json.dumps(ctr_data), 'conversion_data': json.dumps(conversion_data) })

<!-- dashboard.html --> <script src="https://cdn.jsdelivr.net/npm/echarts@5.4.3/dist/echarts.min.js"></script> <script> var chart = echarts.init(document.getElementById('chart')); var option = { tooltip: {trigger: 'axis'}, legend: {data: ['CTR', '转化率']}, xAxis: {type: 'category', data: {{ dates|safe }}}, yAxis: {type: 'value'}, series: [ {name: 'CTR', type: 'line', data: {{ ctr_data|safe }}}, {name: '转化率', type: 'line', data: {{ conversion_data|safe }}} ] }; chart.setOption(option); </script>