AB实验设计与因果推断能力：从原理到工程实践

1次阅读

没有评论

共计 2681 个字符，预计需要花费 7 分钟才能阅读完成。

在数据驱动的决策过程中，AB 测试是验证策略效果的核心工具。但实际落地时常常遇到三类典型问题：

统计功效不足 ：当样本量过小时，可能无法检测到真实的效应差异。例如，点击率提升 1% 的实验可能需要数十万样本才能达到 80% 的统计功效
样本污染 ：用户在多组间交叉（如同时进入实验组和对照组）会导致效应稀释
新奇效应 ：用户对新产品功能的初始好奇会扭曲短期指标（如首日留存率暴涨但后续快速衰减）

传统 A / B 测试的完全随机分组（Complete Randomization）在用户异质性显著时，可能因分组不均衡产生偏差。例如教育类 App 中，付费用户与免费用户对功能改动的反应可能完全不同。

适用场景：用户群体同质性强，无显著分层特征
实现简单，通过哈希用户 ID 分桶即可
缺点：在用户画像差异大时，可能产生分组偏置

按用户特征（如活跃度、消费等级）划分 strata
在每个 strata 内独立进行随机分配
优势：确保关键维度上的组间平衡

示例代码：

# 按用户活跃度分层抽样
df['strata'] = pd.qcut(df['active_days'], q=5)
groups = []
for _, stratum in df.groupby('strata'):
    stratum['group'] = np.random.choice(['control','test'], size=len(stratum))
    groups.append(stratum)
result = pd.concat(groups)

适用于观测数据（无法随机实验时）
通过倾向得分匹配构造可比样本
缺点：依赖强可忽略性假设（no unmeasured confounders）

基本模型：
$$\Delta Y = (Y_{test,after} – Y_{test,before}) – (Y_{control,after} – Y_{control,before})$$

关键假设：平行趋势假设（实验前两组的变化趋势一致）。Python 实现示例：

import statsmodels.formula.api as smf
# 生成时间虚拟变量
did_data['post_treatment'] = (did_data['period'] >= '2023-01-01').astype(int)
did_data['treated'] = did_data['group'] == 'test'
# 拟合 DID 模型
model = smf.ols('metric ~ treated*post_treatment + C(user_strata)', data=did_data).fit()
print(model.summary())

使用逻辑回归估计倾向得分：
$$ e(X) = P(T=1|X) $$

最近邻匹配代码示例：

from sklearn.neighbors import NearestNeighbors
# 计算倾向得分
ps_model = LogisticRegression().fit(covariates, treatment_flag)
ps_scores = ps_model.predict_proba(covariates)[:,1]
# 最近邻匹配
nbrs = NearestNeighbors(n_neighbors=1).fit(ps_scores.reshape(-1,1))
_, indices = nbrs.kneighbors(ps_scores[treatment_flag==0].reshape(-1,1))
matched_control = control_group.iloc[indices.flatten()]

使用统计功效分析确定最小样本量：

from statsmodels.stats.power import TTestIndPower
# 参数设置
effect_size = 0.02  # 预期效应大小
alpha = 0.05        # 显著性水平
power = 0.8         # 统计功效
# 计算样本量
analysis = TTestIndPower()
sample_size = analysis.solve_power(effect_size=effect_size, 
                                  alpha=alpha, 
                                  power=power,
                                  ratio=1.0)
print(f'每组需要样本量: {int(sample_size)}')

应对样本衰减的方法：

生存分析 ：使用 Kaplan-Meier 曲线评估不同时间点的效应
滚动时间窗 ：采用滑动窗口计算指标（如 7 日滚动留存率）
队列分析 ：按用户进入实验的时间分批次比较

典型场景：实验组整体转化率更高，但每个用户分群的转化率都低于对照组。解决方法：

确保分群维度合理（如按用户生命周期阶段）

使用 Mantel-Haenszel 检验进行分层分析

from statsmodels.stats.contingency_tables import StratifiedTable
# 构造分层列联表
table = StratifiedTable.from_data(df['strata'], 
                                df['group'], 
                                df['conversion'])
print(table.summary())

正交分层设计要点：

使用哈希函数确保不同实验层互不干扰

推荐 Google 的分层哈希方案：

def assign_bucket(user_id, salt, num_buckets=100):
    key = f'{user_id}_{salt}'.encode('utf-8')
    hash_val = int(hashlib.md5(key).hexdigest(), 16)
    return hash_val % num_buckets

当实验组和对照组的用户画像存在系统性差异时（如测试组意外包含更多高价值用户），可以考虑：