AI 自动化数据回测：新特征上线前，先回溯三个月表现

发布时间：2026/7/11 20:13:19

AI 自动化数据回测：新特征上线前，先回溯三个月表现

一、新特征上线的"薛定谔恐惧"

我们组每个月至少上线 3-5 个新特征——用户近7天点击序列、商品跨品类关联度、时段偏好权重……每个特征从开发到上线，PM 都会问一句话："这特征靠谱吗？"

以前我的回答是："模型离线 AUC 涨了 0.02。"然后 PM 追问："那线上效果呢？"我只能说："上线看看。"

这就是新特征上线的死循环——想验证得先上线，但不上线又没法验证。直到我们搭了一套 AI 自动化回测系统，把"上线看看"变成了"先回溯三个月数据，模拟真实流量走向"。

flowchart LR A[新特征定义] --> B[AI 自动生成回测方案] B --> C[拉取历史数据] C --> D{模拟流量分配} D -->|流量A| E[对照组<br/>旧特征集合] D -->|流量B| F[实验组<br/>旧特征+新特征] E --> G[计算核心指标差异] F --> G G --> H{显著性检验} H -->|通过| I[AI 生成回测报告] H -->|不通过| J[AI 分析失败原因] style B fill:#7B68EE,color:#fff style I fill:#27AE60,color:#fff style J fill:#E74C3C,color:#fff

二、回测系统的核心设计

回测系统本质上是在用历史数据模拟线上 AB 实验。但这里有个关键难点：历史数据里用户的每一步都是在"旧特征环境"下做出的，你怎么知道"如果当时有新特征，用户会不会点"？

我们的解决方案是反事实模拟（Counterfactual Simulation），分三步走：

第一步：离线特征构建。用 SQL/Python 对历史数据进行回溯计算。比如你要上线"用户近7天购买品类分布"这个特征，那就对过去三个月每一天重新计算一遍——把每一天当天当成"今天"，往前推7天汇总。

第二步：模型影子评分。把新旧两套特征分别灌入模型，得到两组预测分数。这里的关键是保持模型参数不变——因为我们只评估特征贡献，不是评估模型升级。

第三步：指标对比。选一个核心业务指标（比如排序时的 CTR、风控时的拦截率），按时间窗口对比新旧特征下的表现差异。

用代码表达就是：

# backtest_core.py — 回测核心逻辑 import pandas as pd import numpy as np from typing import Dict, List, Tuple from scipy import stats # 用于配对 t 检验 class FeatureBacktester: """新特征回测器：给定新特征定义，在历史数据上模拟 AB 效果""" def __init__(self, model, old_features: List[str], new_feature_name: str): # model: 已训练好的模型（不更新参数，只做推理） self.model = model # 注意：这里期望的 model 有 predict 方法 self.old_features = old_features # 现有特征列表 self.new_feature_name = new_feature_name def build_historical_features(self, raw_data: pd.DataFrame, feature_dates: pd.DatetimeIndex) -> pd.DataFrame: """在历史数据上重新构建特征 这是回测最核心的步骤：用历史数据"假装"回到过去 为什么要按日期回溯？因为要保证训练时不会用到"未来信息" """ # raw_data: 包含用户行为日志的原始数据 # feature_dates: 每天一个时间点，用于构建滑动窗口特征 historical_features = pd.DataFrame(index=feature_dates) # 遍历每一天，构建该天的特征快照 for current_date in feature_dates: # 只看 current_date 之前的数据，避免数据泄露 past_data = raw_data[raw_data['event_date'] < current_date] # 计算近7天点击次数（这是我们的新特征示例） past_7d = past_data[ past_data['event_date'] >= current_date - pd.Timedelta(days=7) ] click_count = past_7d[past_7d['event_type'] == 'click'].shape[0] historical_features.loc[current_date, 'click_7d'] = click_count return historical_features def run_backtest(self, raw_data: pd.DataFrame, feature_dates: pd.DatetimeIndex, metrics: List[str]) -> Dict: """执行完整回测流程，返回对照组 vs 实验组的指标对比""" # 1. 构建新旧两套特征 old_feat = self.build_historical_features(raw_data, feature_dates) # 假设新特征就是多加一列（实际场景可能是多个新特征） new_feat = old_feat.copy() # 2. 模型预测（用旧特征 + 新特征分别预测） old_preds = self.model.predict(old_feat[self.old_features]) new_preds = self.model.predict( pd.concat([old_feat[self.old_features], new_feat[self.new_feature_name]], axis=1) ) # 3. 计算核心指标（以 CTR 提升为例） old_ctr = self._compute_ctr(raw_data, old_preds) new_ctr = self._compute_ctr(raw_data, new_preds) # 4. 显著性检验 — 不能只看均值差异 # 为什么用配对 t 检验？因为对照组和实验组是同一批样本， # 只有特征不同，相互独立性的假设不成立 daily_old = self._compute_daily_ctr(raw_data, old_preds, feature_dates) daily_new = self._compute_daily_ctr(raw_data, new_preds, feature_dates) t_stat, p_value = stats.ttest_rel(daily_new, daily_old) # p_value < 0.05 表示差异显著，不是随机波动 return { "old_ctr": old_ctr, "new_ctr": new_ctr, "ctr_lift": (new_ctr - old_ctr) / old_ctr if old_ctr > 0 else 0, "p_value": p_value, "significant": p_value < 0.05, "daily_breakdown": pd.DataFrame({ "date": feature_dates, "old_ctr": daily_old, "new_ctr": daily_new, "delta": daily_new - daily_old }) } def _compute_ctr(self, data: pd.DataFrame, preds: np.ndarray) -> float: """计算 CTR（点击率）— 这是示例指标，实际可替换为任何业务指标""" # 取预测分 Top 20% 作为"推荐展示位" threshold = np.percentile(preds, 80) # 阈值以上的预测认为会被展示 shown_mask = preds >= threshold if shown_mask.sum() == 0: return 0.0 # 展示的样本中实际发生点击的比例 clicks = data.loc[shown_mask, 'label'].sum() if 'label' in data.columns else 0 return clicks / shown_mask.sum() def _compute_daily_ctr(self, data, preds, dates) -> np.ndarray: """按天拆解 CTR，用于统计检验""" daily_ctr = [] for d in dates: day_mask = data['event_date'] == d if day_mask.sum() > 0: threshold = np.percentile(preds[day_mask], 80) shown = preds[day_mask] >= threshold clicks = data.loc[day_mask & {'label' in data.columns}, 'label'].sum() daily_ctr.append(clicks / max(shown.sum(), 1)) else: daily_ctr.append(np.nan) return np.array(daily_ctr)