基于协同过滤算法的 Flask 理财产品推荐系统实现

理财产品管理

理财订单

反馈信息

产品经理功能界面

核心算法实现

首页理财产品的推荐是系统的亮点，采用协同过滤算法。该算法通过分析用户的历史行为和偏好（如收藏记录），识别用户之间或理财产品之间的相似性。

具体思路是：计算目标用户与其他用户的相似度，找到最相似的用户群体，然后推荐那些相似用户喜欢但目标用户尚未接触过的产品。这里使用的是余弦相似度（Cosine Similarity）来量化向量间的夹角。

代码解析

下面展示了核心的推荐逻辑实现。注意处理 datetime 类型序列化以及 SQL 拼接的安全性问题。

import math
from flask import request, session, jsonify
from sqlalchemy import text

# 计算余弦相似度
# a, b 为两个用户的评分字典 {product_id: rating}
def cosine_similarity(a, b):
    # 分子：对应元素乘积之和
    numerator = sum([a[key] * b[key] for key in a if key in b])
    # 分母：各自模长的乘积
    denominator = math.sqrt(sum([a[key]**2 for key in a])) * math.sqrt(sum([b[key]**2 for key in b]))
    
    if denominator == 0:
        return 0
    return numerator / denominator

# 收藏协同算法接口
@main_bp.route("/pythonlrx30079/licaichanpin/autoSort2", methods=['GET'])
def pythonlrx30079_licaichanpin_autoSort2():
    if request.method == 'GET':
        user_ratings = {}
        req_dict = session.get("req_dict")
        userinfo = session.get("params")
        
        # 查询当前用户的收藏记录
        sql = "select * from storeup where type = 1 and tablename = 'licaichanpin' order by addtime desc"
        data = db.session.execute(sql)
        data_dict = [dict(zip(result.keys(), result)) for result in data.fetchall()]
        
        # 封装 userid、refid 的矩阵
        for item in data_dict:
            uid = str(item["userid"])
            rid = str(item["refid"])
            if uid not in user_ratings:
                user_ratings[uid] = {}
            if rid in user_ratings[uid]:
                user_ratings[uid][rid] += 1
            else:
                user_ratings[uid][rid] = 1

        sorted_recommended_goods = []
        try:
            # 计算目标用户与其他用户的相似度
            target_uid = userinfo.get("id")
            similarities = {
                other_user: cosine_similarity(user_ratings[target_uid], user_ratings[other_user]) 
                for other_user in user_ratings 
                if other_user != target_uid
            }
            
            # 找到与目标用户最相似的用户
            most_similar_user = sorted(similarities, key=similarities.get, reverse=True)[0]
            
            # 找到最相似用户喜欢但目标用户未购买过的商品
            recommended_goods = {
                goods: rating 
                for goods, rating in user_ratings[most_similar_user].items() 
                if goods not in user_ratings[target_uid]
            }
            
            # 按评分降序排列推荐
            sorted_recommended_goods = sorted(recommended_goods, key=recommended_goods.get, reverse=True)
        except Exception:
            pass

        L = []
        # 构建查询条件，优先展示推荐商品
        where_conditions = [
            f"{key} = '{value}'" 
            for key, value in req_dict.items() 
            if key not in ["page", "limit", "order", "sort"]
        ]
        where = " AND ".join(where_conditions) if where_conditions else "1=1"

        # 执行查询：推荐商品优先，其余后置
        if where != "1=1":
            sql = f'''SELECT * FROM (SELECT * FROM licaichanpin WHERE {where}) AS table1 
                      WHERE id IN ('{"','".join(sorted_recommended_goods)}') 
                      UNION ALL 
                      SELECT * FROM (SELECT * FROM licaichanpin WHERE {where}) AS table1 
                      WHERE id NOT IN ('{"','".join(sorted_recommended_goods)}')'''
        else:
            sql = f"select * from licaichanpin where id in ('%s'%('","'.join(sorted_recommended_goods)+") union all select * from licaichanpin where id not in('%s'%('","'.join(sorted_recommended_goods)+")"

        data = db.session.execute(text(sql))
        data_dict = [dict(zip(result.keys(), result)) for result in data.fetchall()]

        # 处理 datetime 对象序列化
        for online_dict in data_dict:
            for key in online_dict:
                val_type = str(type(online_dict[key]))
                if 'datetime.datetime' in val_type or 'datetime' in val_type:
                    online_dict[key] = online_dict[key].strftime("%Y-%m-%d %H:%M:%S")
            L.append(online_dict)

        # 返回封装的 json 结果
        limit_count = int(req_dict.get('limit', 5))
        return jsonify({
            "code": 0, 
            "msg": '', 
            "data": {
                "currPage": 1,
                "totalPage": 1,
                "total": len(L),
                "pageSize": limit_count,
                "list": L[:limit_count]
            }
        })

总结

本系统实现了基于 Flask 的理财产品推荐功能，通过协同过滤算法有效提升了推荐的准确性。虽然在实际应用中还有优化空间（如引入更多特征维度），但整体已达成预期设计目的。开发过程中，细心和耐心是保证代码质量的关键，后续可进一步丰富系统功能以应对更复杂的业务场景。