第23课：推荐数据管道

🎯 推荐数据管道

构建推荐系统数据管道，从用户行为收集到特征工程到训练数据准备。

🐍 Python实战：核心实战

# 推荐数据管道
import random, time
from datetime import datetime
from collections import defaultdict

class UserBehaviorCollector:
    def __init__(self): self.events = []
    def collect(self, user_id, item_id, action, timestamp=None):
        self.events.append({'user_id':user_id,'item_id':item_id,'action':action,'ts':timestamp or datetime.now().isoformat()})
        return len(self.events)

class FeatureEngine:
    def __init__(self): self.user_features = defaultdict(dict); self.item_features = defaultdict(dict)
    def compute_user_features(self, events):
        user_stats = defaultdict(lambda: {'views':0,'clicks':0,'purchases':0,'categories':defaultdict(int)})
        for e in events:
            uid = e['user_id']
            if e['action'] == 'view': user_stats[uid]['views'] += 1
            elif e['action'] == 'click': user_stats[uid]['clicks'] += 1
            elif e['action'] == 'purchase': user_stats[uid]['purchases'] += 1
            user_stats[uid]['categories'][e['item_id'][:3]] += 1
        for uid, stats in user_stats.items():
            self.user_features[uid] = {
                'total_actions': stats['views']+stats['clicks']+stats['purchases'],
                'ctr': round(stats['clicks']/max(stats['views'],1), 3),
                'conv_rate': round(stats['purchases']/max(stats['clicks'],1), 3),
                'top_category': max(stats['categories'], key=stats['categories'].get) if stats['categories'] else 'none',
                'is_active': stats['views'] > 5
            }
    def compute_item_features(self, events):
        item_stats = defaultdict(lambda: {'views':0,'clicks':0,'purchases':0,'revenue':0})
        for e in events:
            iid = e['item_id']
            if e['action'] == 'view': item_stats[iid]['views'] += 1
            elif e['action'] == 'click': item_stats[iid]['clicks'] += 1
            elif e['action'] == 'purchase':
                item_stats[iid]['purchases'] += 1
                item_stats[iid]['revenue'] += random.uniform(10, 200)
        for iid, stats in item_stats.items():
            self.item_features[iid] = {
                'popularity': stats['views'],
                'ctr': round(stats['clicks']/max(stats['views'],1), 3),
                'conv_rate': round(stats['purchases']/max(stats['clicks'],1), 3),
                'total_revenue': round(stats['revenue'], 2)
            }

class SampleGenerator:
    def __init__(self): self.samples = []
    def generate(self, events):
        user_items = defaultdict(set)
        for e in events:
            if e['action'] == 'purchase': user_items[e['user_id']].add(e['item_id'])
        for uid, items in user_items.items():
            for item in items: self.samples.append({'user_id':uid,'item_id':item,'label':1})
            # Negative sampling
            all_items = set(e['item_id'] for e in events)
            neg = list(all_items - items)[:min(3, len(all_items)-len(items))]
            for item in neg: self.samples.append({'user_id':uid,'item_id':item,'label':0})

# 运行
collector = UserBehaviorCollector()
categories = ['CAT_A','CAT_B','CAT_C']
actions = ['view','click','purchase']
for _ in range(2000):
    uid = f"U{random.randint(1,200)}"
    iid = f"{random.choice(categories)}_{random.randint(1,50)}"
    action = random.choices(actions, weights=[0.6,0.3,0.1])[0]
    collector.collect(uid, iid, action)

feature_engine = FeatureEngine()
feature_engine.compute_user_features(collector.events)
feature_engine.compute_item_features(collector.events)

sample_gen = SampleGenerator()
sample_gen.generate(collector.events)

print("推荐数据管道:")
print(f"  行为事件: {len(collector.events)}")
print(f"  用户特征: {len(feature_engine.user_features)}")
print(f"  物品特征: {len(feature_engine.item_features)}")
print(f"  训练样本: {len(sample_gen.samples)} (正例{sum(1 for s in sample_gen.samples if s['label']==1)})")
sample_user = list(feature_engine.user_features.items())[0]
print(f"  用户示例: {sample_user[0]} → {sample_user[1]}")
print("✅ 验证通过 - 推荐数据管道运行正常")

📖 原理深入：企业数据平台架构

🏆 最佳实践：数据平台建设检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答