5.1 电商推荐系统实现（上）
5.1 电商推荐系统实现（上）
本节导读：基于FAISS构建高精度电商推荐系统，通过用户行为分析和商品相似度计算，实现个性化推荐功能，提升用户体验和转化率。
学习目标
- 理解电商推荐系统的核心架构
- 掌握基于FAISS的物品协同过滤实现方法
- 学会用户画像构建和推荐策略优化
- 能够设计和实现高性能推荐引擎
核心概念
电商推荐系统是现代电商平台的核心组件，通过分析用户行为和商品特征，为用户提供个性化的商品推荐。FAISS在电商推荐中主要用于商品相似度计算和快速检索。
推荐系统架构
电商推荐系统通常包含以下层次：
- 数据层：用户行为数据、商品数据、上下文数据
- 特征层：用户特征提取、商品特征向量化
- 算法层：推荐算法、相似度计算、候选生成
- 服务层：推荐接口、实时响应、效果评估
FAISS在推荐中的作用
FAISS在推荐系统中主要用于：
- 商品相似度计算：快速计算商品间的相似度
- 最近邻检索：找到与用户兴趣最相似的商品
- 高效候选生成：从大规模商品库中快速筛选候选集
- 实时推荐：支持低延迟的实时推荐请求
系统架构设计
整体架构
# 电商推荐系统架构 class ECommerceRecommendationSystem: def __init__(self, config): self.config = config self.user_data_manager = UserDataManager(config) self.product_data_manager = ProductDataManager(config) self.feature_extractor = FeatureExtractor(config) self.faiss_index = FAISSIndexManager(config) self.recommendation_engine = RecommendationEngine(config) def build_system(self): """构建完整推荐系统""" print("开始构建电商推荐系统...") # 1. 数据准备 self._prepare_data() # 2. 特征提取 self._extract_features() # 3. 构建FAISS索引 self._build_faiss_index() # 4. 初始化推荐引擎 self._init_recommendation_engine() print("电商推荐系统构建完成！") def _prepare_data(self): """准备数据""" print("准备用户和商品数据...") # 加载用户数据 self.user_data = self.user_data_manager.load_user_data() print(f"加载了 {len(self.user_data)} 个用户数据") # 加载商品数据 self.product_data = self.product_data_manager.load_product_data() print(f"加载了 {len(self.product_data)} 个商品数据") def _extract_features(self): """提取特征""" print("提取用户和商品特征...") # 提取用户特征 self.user_features = self.feature_extractor.extract_user_features(self.user_data) print(f"用户特征维度: {self.user_features.shape}") # 提取商品特征 self.product_features = self.feature_extractor.extract_product_features(self.product_data) print(f"商品特征维度: {self.product_features.shape}") def _build_faiss_index(self): """构建FAISS索引""" print("构建FAISS索引...") # 构建商品索引 self.faiss_index.build_index(self.product_features) print(f"商品索引构建完成，包含 {len(self.product_features)} 个商品") def _init_recommendation_engine(self): """初始化推荐引擎""" print("初始化推荐引擎...") self.recommendation_engine.initialize( user_features=self.user_features, product_features=self.product_features, faiss_index=self.faiss_index ) print("推荐引擎初始化完成")
数据管理模块
class DataManager: def __init__(self, config): self.config = config self.data_source = config['data_source'] def load_data(self): """加载数据""" if self.data_source == 'database': return self._load_from_database() elif self.data_source == 'file': return self._load_from_file() else: raise ValueError(f"未知的数据源: {self.data_source}") class UserDataManager(DataManager): def _load_from_database(self): """从数据库加载用户数据""" # 实现数据库连接和数据加载 # 返回用户行为数据 return self._parse_user_data() def _load_from_file(self): """从文件加载用户数据""" # 实现文件读取 # 返回用户行为数据 return self._parse_user_data() def _parse_user_data(self): """解析用户数据""" # 这里应该返回结构化的用户数据 # 格式: {user_id: {interactions: [...], preferences: {...}}} return {} class ProductDataManager(DataManager): def _load_from_database(self): """从数据库加载商品数据""" # 实现商品数据加载 return self._parse_product_data() def _parse_product_data(self): """解析商品数据""" # 返回商品特征数据 # 格式: {product_id: {features: [...], metadata: {...}}} return {}
用户特征工程
用户特征提取
import numpy as np import pandas as pd from collections import defaultdict from sklearn.preprocessing import StandardScaler class UserFeatureExtractor: def __init__(self, config): self.config = config self.scaler = StandardScaler() def extract_user_features(self, user_data): """提取用户特征""" print("开始提取用户特征...") # 1. 统计用户行为特征 behavioral_features = self._extract_behavioral_features(user_data) # 2. 提取用户偏好特征 preference_features = self._extract_preference_features(user_data) # 3. 提取人口统计学特征 demographic_features = self._extract_demographic_features(user_data) # 4. 合并特征 all_features = self._combine_features( behavioral_features, preference_features, demographic_features ) # 5. 标准化 normalized_features = self._normalize_features(all_features) print(f"用户特征提取完成，特征维度: {normalized_features.shape}") return normalized_features def _extract_behavioral_features(self, user_data): """提取用户行为特征""" behavioral_features = [] for user_id, user_info in user_data.items(): # 行为统计特征 interactions = user_info.get('interactions', []) features = { 'total_interactions': len(interactions), 'purchase_rate': self._calculate_purchase_rate(interactions), 'click_rate': self._calculate_click_rate(interactions), 'cart_rate': self._calculate_cart_rate(interactions), 'view_rate': self._calculate_view_rate(interactions), 'avg_session_duration': self._calculate_avg_session_duration(interactions), 'session_frequency': self._calculate_session_frequency(interactions), 'diversity_index': self._calculate_diversity_index(interactions) } behavioral_features.append(list(features.values())) return np.array(behavioral_features) def _extract_preference_features(self, user_data): """提取用户偏好特征""" preference_features = [] for user_id, user_info in user_data.items(): # 商品类别偏好 category_prefs = self._extract_category_preferences(user_info) # 价格偏好 price_prefs = self._extract_price_preferences(user_info) # 品牌偏好 brand_prefs = self._extract_brand_preferences(user_info) # 时间偏好 time_prefs = self._extract_time_preferences(user_info) # 合并偏好特征 user_preferences = [] user_preferences.extend(category_prefs) user_preferences.extend(price_prefs) user_preferences.extend(brand_prefs) user_preferences.extend(time_prefs) preference_features.append(user_preferences) return np.array(preference_features) def _extract_demographic_features(self, user_data): """提取人口统计学特征""" demographic_features = [] for user_id, user_info in user_data.items(): # 基本人口统计特征 age_group = self._encode_age_group(user_info.get('age', 25)) gender = self._encode_gender(user_info.get('gender', 'unknown')) location = self._encode_location(user_info.get('location', 'unknown')) education = self._encode_education(user_info.get('education', 'unknown')) demo_features = [age_group, gender, location, education] demographic_features.append(demo_features) return np.array(demographic_features) def _calculate_purchase_rate(self, interactions): """计算购买转化率""" if not interactions: return 0.0 purchases = sum(1 for action in interactions if action.get('type') == 'purchase') return purchases / len(interactions) def _calculate_click_rate(self, interactions): """计算点击率""" if not interactions: return 0.0 clicks = sum(1 for action in interactions if action.get('type') == 'click') return clicks / len(interactions) def _calculate_cart_rate(self, interactions): """计算加购率""" if not interactions: return 0.0 carts = sum(1 for action in interactions if action.get('type') == 'cart') return carts / len(interactions) def _calculate_view_rate(self, interactions): """计算浏览率""" if not interactions: return 0.0 views = sum(1 for action in interactions if action.get('type') == 'view') return views / len(interactions) def _extract_category_preferences(self, user_info): """提取用户类别偏好""" interactions = user_info.get('interactions', []) category_counts = defaultdict(int) for action in interactions: category = action.get('category', 'unknown') category_counts[category] += 1 # 获取前N个偏好的类别 top_categories = sorted(category_counts.items(), key=lambda x: x[1], reverse=True)[:5] # 创建偏好向量 category_vector = [0] * 10 # 假设有10个主要类别 for i, (category, count) in enumerate(top_categories): if i < len(category_vector): category_vector[i] = count return category_vector def _extract_price_preferences(self, user_info): """提取价格偏好""" interactions = user_info.get('interactions', []) prices = [] for action in interactions: if action.get('type') == 'purchase': price = action.get('price', 0) prices.append(price) if not prices: return [0, 0, 0, 0] # 平均价格，价格标准差，最低价格，最高价格 avg_price = np.mean(prices) price_std = np.std(prices) min_price = np.min(prices) max_price = np.max(prices) return [avg_price, price_std, min_price, max_price] def _extract_brand_preferences(self, user_info): """提取品牌偏好""" interactions = user_info.get('interactions', []) brand_counts = defaultdict(int) for action in interactions: brand = action.get('brand', 'unknown') brand_counts[brand] += 1 # 获取前N个偏好的品牌 top_brands = sorted(brand_counts.items(), key=lambda x: x[1], reverse=True)[:5] # 创建品牌偏好向量 brand_vector = [0] * 5 for i, (brand, count) in enumerate(top_brands): if i < len(brand_vector): brand_vector[i] = count return brand_vector def _extract_time_preferences(self, user_info): """提取时间偏好""" interactions = user_info.get('interactions', []) hourly_activity = [0] * 24 for action in interactions: timestamp = action.get('timestamp', 0) hour = self._timestamp_to_hour(timestamp) hourly_activity[hour] += 1 return hourly_activity def _encode_age_group(self, age): """编码年龄组""" if age < 18: return 0 elif age < 25: return 1 elif age < 35: return 2 elif age < 45: return 3 elif age < 55: return 4 else: return 5 def _encode_gender(self, gender): """编码性别""" return 1 if gender.lower() == 'male' else 0 def _encode_location(self, location): """编码位置""" # 简化版：基于城市编码 location_hash = hash(location) % 10 return location_hash def _encode_education(self, education): """编码教育程度""" education_map = { 'high_school': 0, 'bachelor': 1, 'master': 2, 'phd': 3, 'other': 4 } return education_map.get(education.lower(), 4) def _calculate_avg_session_duration(self, interactions): """计算平均会话时长""" if not interactions: return 0.0 durations = [] current_session = [] for action in interactions: if action.get('type') == 'session_start': current_session = [] elif action.get('type') == 'session_end' and current_session: duration = sum(action.get('timestamp', 0) - start['timestamp'] for start in current_session) durations.append(duration) current_session = [] else: current_session.append(action) return np.mean(durations) if durations else 0.0 def _calculate_session_frequency(self, interactions): """计算会话频率""" if not interactions: return 0.0 sessions = sum(1 for action in interactions if action.get('type') == 'session_start') total_days = self._get_total_days(interactions) return sessions / total_days if total_days > 0 else 0.0 def _calculate_diversity_index(self, interactions): """计算行为多样性指数""" if not interactions: return 0.0 # 统计不同类型行为的数量 action_types = set(action.get('type') for action in interactions) unique_categories = set(action.get('category') for action in interactions) # 多样性指数 = 唯一行为类型数 / 总行为数 diversity = len(action_types) / len(interactions) category_diversity = len(unique_categories) / len(interactions) if interactions else 0 return (diversity + category_diversity) / 2 def _get_total_days(self, interactions): """获取总天数""" if not interactions: return 1 timestamps = [action.get('timestamp', 0) for action in interactions] min_time = min(timestamps) max_time = max(timestamps) days = (max_time - min_time) / (24 * 3600) return max(days, 1) def _timestamp_to_hour(self, timestamp): """时间戳转换为小时""" return (timestamp // 3600) % 24 def _combine_features(self, behavioral, preference, demographic): """合并所有特征""" # 确保所有特征数组长度一致 min_length = min(len(behavioral), len(preference), len(demographic)) combined = [] for i in range(min_length): user_features = [] user_features.extend(behavioral[i]) user_features.extend(preference[i]) user_features.extend(demographic[i]) combined.append(user_features) return np.array(combined) def _normalize_features(self, features): """标准化特征""" if len(features) == 0: return features return self.scaler.fit_transform(features)
评论区
(0)
U