因果推断导论：从基础概念到Python实战

1次阅读

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

经典的冰淇淋销量与溺水事件案例揭示了传统统计方法的局限性。夏季高温时，冰淇淋销量和溺水事件同时增加，但二者并无直接因果关系——真实原因是气温升高。这种伪相关（spurious correlation）现象在业务分析中极为常见：

电商场景：用户活跃度与广告点击率同步上升，可能是节假日效应而非广告真实效果
医疗领域：服用某种药物的患者康复更快，实际可能是该药物优先开给了健康状况更好的群体

数学上，相关性定义为 $\text{Corr}(X,Y)=\frac{E[(X-\mu_X)(Y-\mu_Y)]}{\sigma_X\sigma_Y}$，而因果效应需要满足反事实条件 $E[Y(1)-Y(0)|T=1]$，其中 $Y(1)$ 和 $Y(0)$ 分别表示处理组和对照组的潜在结果。

Rubin 因果模型用数学形式定义个体处理效应（ITE）：
$$\text{ITE}_i = Y_i(1) – Y_i(0)$$
由于现实中无法同时观测到两种状态，我们通常估计平均处理效应（ATE）：
$$\text{ATE} = E[Y(1) – Y(0)]$$
对于已处理群体，关注处理组平均处理效应（ATT）：
$$\text{ATT} = E[Y(1) – Y(0)|T=1]$$

稳定单元处理值假设（SUTVA）：个体的结果不受其他单元处理的干扰
可忽略性假设（Ignorability）：给定协变量 X 后，处理分配 T 与潜在结果独立
重叠性假设（Overlap）：每个个体都有接受处理 / 对照的非零概率

工具库	核心优势	典型场景
DoWhy	端到端因果推理框架	快速验证因果假设的完整性
EconML	基于机器学习的方法	异质性处理效应估计
CausalML	树模型扩展	Uplift 建模
PyWhy	因果发现算法	未知因果结构探索

用因果图明确变量间关系，识别混淆变量（confounder）、中介变量（mediator）和碰撞变量（collider）：

from dowhy import CausalModel
import networkx as nx

model = CausalModel(
    data=df,
    treatment='ad_exposure',
    outcome='purchase',
    graph="""digraph {
        user_activity -> ad_exposure;
        user_activity -> purchase;
        ad_exposure -> purchase;
        season -> user_activity;
        season -> ad_exposure;
    }"""
)
model.view_model()

将因果问题转化为统计估计问题，验证是否可计算 ATE：

identified_estimand = model.identify_effect(proceed_when_unidentifiable=True)
print(identified_estimand)

应用不同估计方法，以倾向得分匹配为例：

from sklearn.linear_model import LogisticRegression

psm_estimate = model.estimate_effect(
    identified_estimand,
    method_name='backdoor.propensity_score_matching',
    target_units='ate',
    method_params={'propensity_score_model': LogisticRegression()}
)
print(psm_estimate)

通过安慰剂测试等验证结果稳健性：

refute_placebo = model.refute_estimate(
    identified_estimand, 
    psm_estimate,
    method_name='placebo_treatment_refuter',
    placebo_type='permute'
)
print(refute_placebo)

未测量混淆变量 ：
现象：估计效应包含隐藏偏差
解法：进行敏感性分析（如 E 值计算）
过度控制中介变量 ：
现象：阻断因果路径导致低估真实效应
解法：使用前门准则（front-door criterion）
样本重叠不足 ：
现象：倾向得分极端值导致匹配质量差
解法：设置 caliper 限制或进行权重修剪

如何设计 A / B 测试验证因果结论？需考虑：
随机化单元的选择（用户级 vs 会话级）
最小可检测效应（MDE）计算
新奇效应（Novelty Effect）的控制
当无法进行随机实验时：
断点回归（RDD）在运营策略评估中的应用
双重差分法（DID）处理面板数据
工具变量（IV）解决内生性问题
因果发现算法：
PC 算法在未知因果结构中的应用
非线性因果关系的约束优化方法

# 生成模拟数据
from dowhy.datasets import linear_dataset
import pandas as pd

np.random.seed(42)
data_dict = linear_dataset(
    beta=5,  # 真实因果效应
    num_common_causes=3,
    num_samples=2000,
    treatment_is_binary=True
)
df = data_dict['df']

# 完整分析流程
model = CausalModel(
    data=df,
    treatment=data_dict['treatment_name'],
    outcome=data_dict['outcome_name'],
    graph=data_dict['gml_graph']
)

# 可视化因果图
model.view_model(layout='dot')

# 识别因果效应
identified_estimand = model.identify_effect()

# 双重机器学习估计
from sklearn.ensemble import GradientBoostingRegressor
dml_estimate = model.estimate_effect(
    identified_estimand,
    method_name='backdoor.econml.dml.DML',
    control_value=0,
    treatment_value=1,
    target_units='ate',
    method_params={
        'init_params': {'model_y': GradientBoostingRegressor(),
            'model_t': GradientBoostingRegressor(),
            'model_final': GradientBoostingRegressor()},
        'fit_params': {}}
)
print(f"DML 估计结果: {dml_estimate.value:.4f}")

# 鲁棒性验证
ref_random = model.refute_estimate(
    identified_estimand,
    dml_estimate,
    method_name='random_common_cause'
)
print(ref_random)