from database import RedisDatabaseHelper, MySqlDao import pandas as pd from models import UserItemScore, SimilarityMatrix import numpy as np from tqdm import tqdm from scipy.sparse import csr_matrix from joblib import Parallel, delayed class ItemCFModel: def __init__(self): self._recommendations = {} self._dao = MySqlDao() def train(self, city_uuid, n=300, k=100, top_n=300, n_jobs=4): # self._score_df = pd.read_csv(score_path) # self._similarity_df = pd.read_csv(similatity_path, index_col=0) print("itemcf: 正在加载order_info...") self._order_data = self._dao.load_order_data(city_uuid) print("正在计算品规培育分数...") self._score_df = UserItemScore(self._order_data).generate_product_scores() print("正在计算商户相似度矩阵...") self._similarity_df = SimilarityMatrix(self._order_data).generate_similarity_matrix() similarity_matrix = csr_matrix(self._similarity_df.values) shop_index = {shop: idx for idx, shop in enumerate(self._similarity_df.index)} index_shop = {idx: shop for idx, shop in enumerate(self._similarity_df.index)} def process_product(product_code, scores): # 获取热度最高的n个商户 top_n_shops = scores.nlargest(n, "score")["cust_code"].values top_n_indices = [shop_index[shop] for shop in top_n_shops] # 找到每个商户最相似的k个商户 similar_shops = {} for shop_idx in top_n_indices: similarities = similarity_matrix[shop_idx].toarray().flatten() similar_indices = np.argpartition(similarities, -k-1)[-k-1:] similar_indices = similar_indices[similar_indices != shop_idx][:k] similar_shops[index_shop[shop_idx]] = [index_shop[idx] for idx in similar_indices] # 生成候选商户列表 candidate_shops = list(set(top_n_shops).union(set([m for sublist in similar_shops.values() for m in sublist]))) candidate_indices = [shop_index[shop] for shop in candidate_shops] # 计算每个候选商户的兴趣得分 interest_scores = {} for candidate_idx in candidate_indices: interest_score = 0 for shop_idx in top_n_indices: if index_shop[candidate_idx] in similar_shops[index_shop[shop_idx]]: shop_score = scores[scores["cust_code"]==index_shop[shop_idx]]["score"].values[0] interest_score += shop_score * similarity_matrix[shop_idx, candidate_idx] interest_scores[index_shop[candidate_idx]] = interest_score # 将候选商户的兴趣得分转换为字典列表,并按照从大到小排列 sorted_candidates = sorted([{shop_id: s} for shop_id, s in interest_scores.items()], key=lambda x: list(x.values())[0], reverse=True)[:top_n] return product_code, sorted_candidates # 并行处理每个品规 results = Parallel(n_jobs=n_jobs)(delayed(process_product)(product_code, scores) for product_code, scores in tqdm(self._score_df.groupby("product_code"), desc="train:正在计算候选得分")) # 存储结果 self._recommendations = {product_code: sorted_candidates for product_code, sorted_candidates in results} self.to_redis_zset(city_uuid) def to_redis_zset(self, city_uuid): """ 将 self._recommendations 中的数据保存到 Redis 的 Sorted Set (ZSET) 中 存储格式为 fc:product_code,其中商户 ID 作为成员,得分作为分数 """ redis_db = RedisDatabaseHelper() # 存redis之前,先进行删除操作 pattern = f"fc:{city_uuid}:*" keys_to_delete = redis_db.redis.keys(pattern) if keys_to_delete: redis_db.redis.delete(*keys_to_delete) for product_code, recommendations in tqdm(self._recommendations.items(), desc="train:正在存储推荐结果"): redis_key = f"fc:{city_uuid}:{product_code}" zset_data = {} for rec in recommendations: for shop_id, score in rec.items(): try: zset_data[shop_id] = float(score) except ValueError as e: print(f"Error converting score to float for shop_id {shop_id}: {score}") raise e redis_db.redis.zadd(redis_key, zset_data) if __name__ == "__main__": itemcf_model = ItemCFModel() itemcf_model.train("00000000000000000000000011445301", n_jobs=4)