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

核心算法实现

首页的理财产品推荐采用了协同过滤算法。该算法的核心在于分析用户之间的相似性，从而预测用户对未接触产品的兴趣。

在代码层面，我们首先定义了余弦相似度计算函数 cosine_similarity。这一步很关键，它决定了我们如何衡量两个用户向量在多维空间中的接近程度。接着，在主路由中获取当前用户的收藏记录，构建用户 - 商品评分矩阵。

这里有个细节需要注意：我们优先查询目标用户已收藏或关注过的同类型产品，确保基础推荐的准确性。然后计算目标用户与其他所有用户的相似度，找出最相似的那一位（Most Similar User）。最后，从这位相似用户喜欢的产品中，剔除掉目标用户已经购买或收藏过的，剩下的就是潜在的推荐项。

以下是具体的实现逻辑，包含相似度计算和 SQL 查询拼接部分：

# 计算余弦相似度
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]))
    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:
            if user_ratings.__contains__(item["userid"]):
                ratings_dict = user_ratings[item["userid"]]
                if ratings_dict.__contains__(item["refid"]):
                    ratings_dict[str(item["refid"])] += 1
                else:
                    ratings_dict[str(item["refid"])] = 1
            else:
                user_ratings[item["userid"]] = { str(item["refid"]): 1 }
        
        sorted_recommended_goods = []
        try:
            # 计算目标用户与其他用户的相似度
            similarities = {
                other_user: cosine_similarity(user_ratings[userinfo.get("id")], user_ratings[other_user]) 
                for other_user in user_ratings if other_user != userinfo.get("id")
            }
            # 找到最相似的用户
            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[userinfo.get("id")]
            }
            # 按评分降序排列推荐
            sorted_recommended_goods = sorted(recommended_goods, key=recommended_goods.get, reverse=True)
        except:
            pass
        
        L = []
        # 按评分顺序查询要推荐列表 (当前用户收藏关注过的同类型优先)
        where = " AND ".join([
            f"{key} = '{value}'" 
            for key, value in req_dict.items() 
            if key != "page" and key != "limit" and key != "order" and key != "sort"
        ])
        
        if where:
            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 = "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(sql)
        data_dict = [dict(zip(result.keys(), result)) for result in data.fetchall()]
        
        for online_dict in data_dict:
            for key in online_dict:
                if 'datetime.datetime' in str(type(online_dict[key])):
                    online_dict[key] = online_dict[key].strftime("%Y-%m-%d %H:%M:%S")
                elif 'datetime' in str(type(online_dict[key])):
                    online_dict[key] = online_dict[key].strftime("%Y-%m-%d %H:%M:%S")
                else:
                    pass
            L.append(online_dict)
        
        return jsonify({"code": 0, "msg": '', "data":{"currPage":1,"totalPage":1,"total":1,"pageSize":5,"list": L[0:int(req_dict['limit'])]}})

系统测试与总结

在程序上线前，测试环节至关重要。通过多轮功能验证，确保了推荐逻辑的稳定性和数据交互的准确性。尽管在实际开发中难免遇到边界情况，但经过反复调试，系统已能稳定运行。

本项目基于 Flask 与 Python 构建了理财产品推荐系统，核心采用协同过滤算法实现个性化推荐。系统涵盖用户管理、产品发布、订单处理及反馈模块，支持管理员与产品经理的多角色操作。后端使用 MySQL 存储数据，前端交互流畅，实现了从需求分析到系统测试的完整开发流程，验证了算法在实际业务场景中的可行性。后续可根据实际业务量引入更复杂的特征工程或分布式计算优化性能。