AB Testing与因果推断实战：如何避免数据偏差并提升实验效果

1次阅读

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

在理想情况下，AB 测试通过随机分组确保实验组和对照组的可比性。然而在实际业务中，完全随机分组往往难以实现：

选择偏差：用户自主选择进入实验组（如主动领取优惠券），导致两组用户特征分布不均
辛普森悖罗：在细分群体中表现相反的趋势，如新老用户对促销活动的响应差异被整体数据掩盖
时间效应混杂：活动期间的外部因素（如节假日）干扰效果归因

这些局限性使得传统平均处理效应 (ATE) 估计失效，此时需要因果推断方法来纠正偏差。

方法	适用场景	实现复杂度	关键假设
双重差分法(DID)	面板数据，有时间维度对比	中等	平行趋势假设
倾向得分匹配(PSM)	横截面数据，协变量多	较高	条件独立假设
工具变量(IV)	存在测量误差或部分合规	高	排他性约束
断点回归(RDD)	有清晰分界阈值	低	局部连续性

import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression

# 生成模拟数据
np.random.seed(42)
N = 10000
data = pd.DataFrame({'age': np.random.normal(35, 5, N),
    'gender': np.random.binomial(1, 0.5, N),
    'history_purchase': np.random.exponential(100, N),
    'treatment': np.random.binomial(1, 0.3, N),  # 30% 用户获得优惠券
    'week': np.random.randint(1, 5, N)  # 观察周期
})

# 生成结果变量（带混杂效应）data['post_spend'] = 50 + 0.8*data['age'] - 5*data['gender'] + \
                    0.1*data['history_purchase'] + \
                    15*data['treatment'] + np.random.normal(0, 10, N)

from sklearn.neighbors import NearestNeighbors

# 计算倾向得分
ps_model = LogisticRegression()
ps_model.fit(data[['age', 'gender', 'history_purchase']], 
             data['treatment'])
data['ps_score'] = ps_model.predict_proba(data[['age', 'gender', 'history_purchase']])[:,1]

# 最近邻匹配
treated = data[data.treatment==1]
control = data[data.treatment==0]

nn = NearestNeighbors(n_neighbors=1)
nn.fit(control[['ps_score']])

_, indices = nn.kneighbors(treated[['ps_score']])
matched_control = control.iloc[indices.flatten()]

# 平衡性检验
print('Age 均值差异:', 
      treated['age'].mean() - matched_control['age'].mean())

import statsmodels.formula.api as smf

did_data = pd.concat([treated, matched_control])
did_model = smf.ols(
    "post_spend ~ treatment*week", 
    data=did_data).fit()
print(did_model.summary())

平行趋势假设验证
错误做法：直接使用 DID 而未检查实验前趋势
正确方案：绘制实验前各期结果变量的时间趋势图
协变量选择不当
错误做法：仅使用可见特征而忽略潜在混淆因素
正确方案：通过因果图 (DAG) 识别关键协变量
样本量不足
错误做法：PSM 后剩余样本过少导致统计功效不足
正确方案：使用有放回匹配或核密度匹配

分层抽样：对关键维度（如用户等级）预先分层再匹配

分布式计算：使用 PySpark 实现大规模 PSM

from pyspark.sql.functions import col

# 在 Spark 中实现分布式匹配
matched_df = spark.sql("""
  SELECT t.*, c.control_id
  FROM treated t
  JOIN control c
  ON ABS(t.ps_score - c.ps_score) < 0.01
  QUALIFY ROW_NUMBER() OVER(
    PARTITION BY t.user_id 
    ORDER BY ABS(t.ps_score - c.ps_score)) = 1
""")