AB Test中的因果推断：从统计原理到工程实践

1次阅读

没有评论

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

在理想情况下，AB 测试通过随机分组消除混杂变量的影响。但实际业务场景中常遇到以下问题：

用户自选择偏差（Self-selection Bias）：用户可能主动选择进入实验组
非随机分流：运营策略无法实现完全随机化分组
历史效应：实验前后存在外部因素干扰

这些情况会导致传统 AB 测试的转化率对比失效。根据 Rubin(1974) 的潜在结果框架，此时需要因果推断技术估计处理效应（ATE）。

适用场景：
– 面板数据（同一对象实验前后对比）
– 满足平行趋势假设

适用场景：
– 横截面数据
– 协变量维度较高时
– 处理组对照组重叠区域较大

适用场景：
– 个体级干预评估
– 对照组样本量极少时

方法选择决策树：

graph TD
    A[数据是否为面板数据?] -->| 是 | B[DID]
    A -->| 否 | C[样本是否极度不平衡?]
    C -->| 是 | D[合成控制]
    C -->| 否 | E[PSM]

import pandas as pd
from sklearn.preprocessing import StandardScaler

# 读取实验数据（假设包含 treatment 标志和特征列）df = pd.read_csv('ab_test_data.csv')

# 标准化连续变量
cont_cols = ['age', 'income']
scaler = StandardScaler()
df[cont_cols] = scaler.fit_transform(df[cont_cols])

# 处理分类变量
df = pd.get_dummies(df, columns=['city_level'])

from sklearn.linear_model import LogisticRegression

# 划分特征和标签
X = df.drop(['user_id','treatment'], axis=1)
y = df['treatment']

# 训练逻辑回归模型
ps_model = LogisticRegression(penalty='l2', C=1.0)
ps_model.fit(X, y)

# 获取倾向得分
df['propensity_score'] = ps_model.predict_proba(X)[:,1]

from sklearn.neighbors import NearestNeighbors

# 划分处理组和对照组
treat_df = df[df.treatment==1]
control_df = df[df.treatment==0]

# 最近邻匹配（1:1）nn = NearestNeighbors(n_neighbors=1, algorithm='ball_tree')
nn.fit(control_df[['propensity_score']])

# 为每个处理组样本找到最近邻
distances, indices = nn.kneighbors(treat_df[['propensity_score']])

# 构建匹配后数据集
matched_pairs = pd.concat([treat_df.reset_index(drop=True),
    control_df.iloc[indices.flatten()].reset_index(drop=True)
], axis=1)

面对百万级样本时，推荐以下优化方案：

分层抽样：先对对照组进行分层抽样减少匹配基数
近似最近邻：使用 Annoy 或 Faiss 加速向量搜索

分布式计算：

# 使用 PySpark 实现分布式 PSM
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(inputCols=feature_cols, outputCol="features")
lr = LogisticRegression(featuresCol="features", labelCol="treatment")