AB测试中的因果推断：从入门到实战避坑指南

1次阅读

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

刚接触 AB 测试时，很容易陷入一个误区：把实验组和对照组的均值差异直接当作处理效应。但现实中至少有三个隐藏陷阱：

用户自选择偏差：比如测试新功能时，活跃用户更可能主动切换到实验组，导致组间基线不平衡
外部因素干扰：促销活动期间突然上涨的指标，可能被误判为产品改进的效果
样本污染：用户在不同组间反复横跳（常见于非强制更新的客户端实验）

去年我们有个惨痛教训：某推荐算法迭代的 AB 测试显示点击率提升 12%，全量上线后反而下跌。后来发现是实验期间恰逢节假日，对照组中包含大量低活用户。

方法	适用场景	最小样本量要求	并行实验支持	学习成本
双重差分法(DID)	存在明确政策冲击时点	每组 >500	★★☆☆☆	低
倾向得分匹配(PSM)	非随机分组但特征观测完整	对照组 >1000	★★★☆☆	中
合成控制法	实验组极少（如城市级实验）	对照组 >50	★☆☆☆☆	高

注：样本量指匹配 / 加权后的有效样本，实际需要更大原始样本

# 安装因果推断工具包
!pip install causalml==0.9.0
# 类型标注让代码更健壮
from typing import Tuple, Dict
import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression

计算倾向得分

def calculate_ps(df: pd.DataFrame, features: list) -> np.ndarray:
    """
    用逻辑回归估计倾向得分
    :param df: 包含特征和 treatment 标志的 DataFrame
    :param features: 用于平衡的协变量列表
    :return: 倾向得分数组
    """
    lr = LogisticRegression(max_iter=1000)
    lr.fit(df[features], df['treatment'])
    return lr.predict_proba(df[features])[:, 1]  # WHY: 用预测概率而非类别

最近邻匹配（带卡钳）

def psm_match(
    df: pd.DataFrame, 
    ps_col: str,
    caliper: float = 0.2
) -> Tuple[pd.DataFrame, Dict[str, float]]:
    """
    执行 1:1 最近邻匹配
    :param caliper: 最大允许 PS 差异（标准差单位）:return: (匹配后数据, 平衡性报告)
    """
    treated = df[df.treatment==1].copy()
    control = df[df.treatment==0].copy()

    # 动态计算卡钳值
    ps_std = np.std(df[ps_col])
    max_dist = caliper * ps_std  # WHY: 避免匹配质量过差

    matched_pairs = []
    for _, treat_row in treated.iterrows():
        # 计算所有对照样本的距离
        distances = np.abs(control[ps_col] - treat_row[ps_col])
        min_idx = distances.idxmin()

        if distances[min_idx] <= max_dist:
            matched_pairs.append((treat_row.name, min_idx))
            control.drop(min_idx, inplace=True)  # 不放回抽样

    # 拼接匹配样本            
    matched_df = pd.concat([df.loc[[i[0] for i in matched_pairs]],
        df.loc[[i[1] for i in matched_pairs]]
    ])

    return matched_df, check_balance(matched_df)

平衡性检查

def check_balance(df: pd.DataFrame) -> Dict[str, float]:
    """检查匹配后特征标准化差异（应 <0.1）"""
    report = {}
    for col in ['age', 'activity_score']:  # 示例特征
        treat_mean = df[df.treatment==1][col].mean()
        control_mean = df[df.treatment==0][col].mean()
        pooled_std = np.sqrt((df[df.treatment==1][col].var() + df[df.treatment==0][col].var()) / 2
        )
        report[col] = abs(treat_mean - control_mean) / pooled_std
    return report

效果评估

# 匹配后直接用 t 检验可能高估显著性
from statsmodels.stats.weightstats import CompareMeans

def treatment_effect(matched_df: pd.DataFrame) -> float:
    """用加权 t 检验计算处理效应"""
    cm = CompareMeans.from_data(matched_df[matched_df.treatment==1]['outcome'],
        matched_df[matched_df.treatment==0]['outcome']
    )
    return cm.ttest_ind(usevar='unequal')[0]  # Welch's t-test

当原始样本不足时，可以：

对匹配后的样本进行 Bootstrap 重采样（至少 1000 次）
使用贝叶斯 PSM 方法（如 causalml 的BART模块）

from causalml.inference.meta import LRSRegressor

lr = LRSRegressor()
te = lr.estimate_ate(X, treatment, y)  # 自动处理小样本

用 Wald 检验判断分组是否真正随机：

from statsmodels.regression.linear_model import OLS

# 检验特征与 treatment 的相关性
model = OLS(df['treatment'], df[features]).fit()
print(model.f_test("features = 0"))  # p>0.1 才安全

同时跑多个实验时，需调整显著性阈值：

from statsmodels.stats.multitest import multipletests

# 原始 p 值列表
pvals = [0.01, 0.04, 0.002] 
adjusted = multipletests(pvals, method='bonferroni')[1]  # [0.03, 0.12, 0.006]