主题
第12章 端到端实战
12.1 概述
前面 11 章分别介绍了系统的每个模块。本章把它们串起来，完成从数据生成到模型训练到预算分配的端到端流程。你会看到整个系统是如何协同工作的，也会学到如何解读分配结果。
12.2 完整流程
mermaid
graph LR
A[生成模拟数据] --> B[构建特征]
B --> C[训练 4 个模型]
C --> D[加载模型 + 预测]
D --> E[AB 分组]
E --> F[预算优化分配]
F --> G[输出结果]三个脚本对应三个阶段：
| 阶段 | 脚本 | 输入 | 输出 |
|---|---|---|---|
| 数据生成 | generate_sample_data.py | 参数（用户数、种子） | data/*.csv |
| 模型训练 | train_models.py | data/*.csv | models/*.pkl |
| 预算分配 | run_allocation.py | data/*.csv + models/*.pkl | data/allocation_result.csv |
12.3 Step 1：生成数据
bash
python scripts/generate_sample_data.py --n-users 5000 --output-dir data生成 5000 个模拟用户...
→ data/users.csv (5000 rows)
生成浏览日志...
→ data/browse_logs.csv (~150,000 rows)
生成AB实验数据...
→ data/experiment_data.csv (5000 rows)
生成价格敏感度标签...
→ data/sensitivity_labels.csv (5000 rows)
✅ 数据生成完成!这里用 5000 个用户做演示（生产环境通常是几百万）。浏览日志约 15 万条，人均 30 条浏览记录。
12.4 Step 2：训练模型
python
<!-- scripts/train_models.py -->
def main():
# 1. 加载数据
users = pd.read_csv("data/users.csv", parse_dates=["register_time"])
browse_logs = pd.read_csv("data/browse_logs.csv", parse_dates=["timestamp"])
experiment = pd.read_csv("data/experiment_data.csv")
sensitivity_labels = pd.read_csv("data/sensitivity_labels.csv")
# 2. 特征工程
pipeline = FeaturePipeline()
features = pipeline.build_features(browse_logs, users, pd.Timestamp.now())
feature_cols = [c for c in pipeline.get_feature_columns()
if c in features.columns]
X = pipeline.encode_categoricals(features[feature_cols])
# 3. 合并标签
merged = features[["user_id"]].merge(experiment, on="user_id", how="inner")
# 4. 训练四个模型
# 4.1 意图预测模型
intent_model = IntentModel()
intent_metrics = intent_model.train(X_train, y_intent)
intent_model.save("models/intent_model.pkl")
# 4.2 Uplift 增量模型
uplift_model = UpliftModel()
uplift_metrics = uplift_model.train(X_train, y_uplift, treatment)
uplift_model.save("models/uplift_model.pkl")
# 4.3 价格敏感度模型
sensitivity_model = SensitivityModel(LGBMParams(objective="regression"))
sens_metrics = sensitivity_model.train(X_train, y_sensitivity)
sensitivity_model.save("models/sensitivity_model.pkl")
# 4.4 GMV 预估模型（只用有订单的样本）
gmv_model = GMVModel(LGBMParams(objective="regression", n_estimators=300))
gmv_metrics = gmv_model.train(X_gmv, y_gmv)
gmv_model.save("models/gmv_model.pkl")运行：
bash
python scripts/train_models.py --data-dir data --model-dir models加载数据...
构建特征...
特征维度: (5000, 30)
训练意图预测模型...
指标: {'auc': 0.72, 'logloss': 0.31}
→ models/intent_model.pkl
训练Uplift增量模型...
指标: {'treatment_metrics': {'auc': 0.71}, 'control_metrics': {'auc': 0.68}}
→ models/uplift_model.pkl
训练价格敏感度模型...
指标: {'mse': 0.012, 'mae': 0.085}
→ models/sensitivity_model.pkl
训练GMV预估模型...
指标: {'mse': 1245.3, 'mae': 25.6}
→ models/gmv_model.pkl
✅ 模型训练完成!指标解读
| 模型 | 核心指标 | 预期范围 | 说明 |
|---|---|---|---|
| 意图模型 | AUC 0.70-0.75 | 好 | 新用户数据有限，0.72 已经不错 |
| Uplift 模型 | Treatment AUC 0.70+ | 好 | 两个子模型的 AUC 都要达标 |
| 敏感度模型 | MAE 0.08-0.10 | 好 | 标签本身是启发式构造的 |
| GMV 模型 | — | — | GMV 预估天然方差大，关注方向对即可 |
12.5 Step 3：运行预算分配
python
<!-- scripts/run_allocation.py -->
def main():
# 加载数据和模型
users = pd.read_csv("data/users.csv")
browse_logs = pd.read_csv("data/browse_logs.csv")
# 特征工程
pipeline = FeaturePipeline()
features = pipeline.build_features(browse_logs, users, snapshot_time)
X = pipeline.encode_categoricals(features[feature_cols])
# 加载模型
intent_model = IntentModel.load("models/intent_model.pkl")
uplift_model = UpliftModel.load("models/uplift_model.pkl")
sensitivity_model = SensitivityModel.load("models/sensitivity_model.pkl")
gmv_model = GMVModel.load("models/gmv_model.pkl")
# AB 分组
ab_test = ABTestFramework(config)
groups = ab_test.assign_groups_batch(features["user_id"].tolist())
# 模型预测
uplift_scores = uplift_model.predict_uplift(X)
predicted_gmv = gmv_model.predict(X)
sensitivity_scores = sensitivity_model.predict(X)
# 分配
allocator = CouponAllocator(config)
# 模型组 → 模型分配
model_allocation = allocator.allocate(
user_features=model_features,
uplift_scores=uplift_scores[model_mask],
predicted_gmv=predicted_gmv[model_mask],
sensitivity_scores=sensitivity_scores[model_mask],
)
# 规则组 → 规则分配
rule_allocation = allocator.allocate_by_rules(rule_features)运行：
bash
python scripts/run_allocation.py --data-dir data --model-dir models加载数据...
构建特征...
AB 分组...
模型预测...
预算优化分配...
分配结果:
总用户: 4991
发券用户: 3705
预估券面额总成本: ¥25,635
平均面额: ¥6.9
分组分布:
model 2250
rule 1500
explore 750
control 49112.6 结果分析
分组分布
control: 10% → 491 人（不发券，对照组）
model: 45% → 2250 人（模型策略）
rule: 30% → 1500 人（规则策略）
explore: 15% → 750 人（探索组）分组比例与 CouponConfig 中的设置一致（10% / 45% / 30% / 15%），验证了 AB 分组的正确性。
发券率
发券用户 3705 / 总用户 4991 = 74.2%
未发券的用户包括：
- 对照组 491 人（不发券）
- 模型组中 uplift_score ≤ 0 的用户（模型判断发券无增量效果）
- 探索组 750 人（当前实现中探索组不发券，只收集数据）
平均面额
平均面额 ¥6.9，远低于最大面额 ¥50。这说明预算优化器在控制成本：大部分用户只需要一张小面额券就够了，高面额券只发给高敏感度的 Persuadable 用户。
预算使用率
预估总成本 ¥25,635，相对于日预算 ¥1,000,000 只用了 2.6%。这是因为演示只用了 5000 个用户。如果扩展到 50 万用户，预算使用率会大幅上升。
12.7 结果文件格式
allocation_result.csv 的格式：
user_id,coupon_amount,expected_cost,expected_gain,efficiency,group
u_000123,10,1.23,4.56,3.71,model
u_000456,15,2.34,5.67,2.42,model
u_000789,20,0.00,0.00,0.00,rule
u_001000,0,,,, control| 字段 | 说明 |
|---|---|
user_id | 用户 ID |
coupon_amount | 分配的面额（0 = 不发） |
expected_cost | 预期成本 = 面额 × 增量转化概率 |
expected_gain | 预期收益 = 增量概率 × 预测 GMV × 毛利率 |
efficiency | 边际效率 = 收益 / 成本 |
group | AB 分组 |
12.8 扩展到生产环境
从演示到生产，需要关注几个点：
| 维度 | 演示环境 | 生产环境 |
|---|---|---|
| 数据规模 | 5,000 用户 | 500 万+ 用户 |
| 特征计算 | Pandas（单机） | Spark / Flink |
| 模型训练 | 单机 LightGBM | 分布式 LightGBM |
| 预算分配 | 内存贪心 | 拉格朗日松弛（可并行） |
| 调度 | 手动脚本 | Airflow DAG |
| 存储 | CSV 文件 | Hive / MySQL / Redis |
拉格朗日松弛方法的一大优势在此体现：给定 λ 后，每个用户的面额选择是独立的，天然支持 MapReduce 并行化。50 万用户在 Spark 上分钟级完成。
本章小结
- 端到端流程：
generate_sample_data.py→train_models.py→run_allocation.py - 四个模型在 5000 个用户上训练正常，指标在合理范围内
- 模型组 74% 的用户被分配了券，平均面额 ¥6.9
- 预算优化器有效控制了成本，避免了过度发券
- 生产环境需要关注数据规模、分布式计算和调度编排
思考题
- 如果模型训练数据中 treatment/control 比例不是 50/50（比如 70/30），对 Uplift 模型会有什么影响？需要做什么调整？
- 分配结果中
efficiency最高的用户有什么特征？这些用户对业务意味着什么？ - 如何设计一个自动化的日度 pipeline，�数据生成 → 模型更新 → 预算分配每天自动运行？