因果推断新范式：2020年后主流算法与框架实战指南

1次阅读

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

传统机器学习模型擅长发现数据中的 关联关系 （Correlation），但在医疗效果评估、广告转化归因等场景中，我们更需要识别 因果效应（Causal Effect）。举个典型例子：

在药品试验中，服药组康复率比对照组高 20%，但这可能只是因为年轻患者更倾向于服药（混杂因素），而非药物本身的效果
电商平台发放优惠券后销量上涨，但需要区分是券的真实作用，还是同期节日流量的影响

这类问题需要 反事实推理（Counterfactual Inference）：假设同一个体同时存在「接受干预」和「未接受干预」两种状态，其结果差异才是真正的因果效应。传统方法如回归分析常因忽略混杂变量（Confounder）导致估计偏差。

核心思想：通过机器学习模型分阶段估计倾向得分（Propensity Score）和结果模型，再用 Neyman 正交化消除偏差

优势：
对模型误设（Misspecification）更鲁棒
支持连续 / 离散处理变量
局限性：
需要样本量较大（>10k）
高维数据需配合特征选择

时间复杂度：𝑂(𝑛⋅(𝑚₁+𝑚₂))，其中𝑚₁,𝑚₂为两个阶段模型复杂度

特色算法：
– DeepIV：用神经网络建模工具变量（Instrumental Variable）
– DRLearner：双重稳健（Doubly Robust）估计器

适用场景：
– 存在未观测混杂（Hidden Confounding）时
– 处理变量为连续值（如剂量响应）

数据假设：需要有效工具变量满足排他性约束（Exclusion Restriction）

明星功能：
– 因果森林（Causal Forest）：基于广义随机森林的异质性处理效应估计
– Meta-Learners：如 X -Learner 处理实验组 / 对照组样本量不平衡

计算效率：基于 LightGBM 实现，适合中等规模数据（1k~100k 样本）

import numpy as np
from econml.iv.dml import DMLIV

# 生成模拟数据（工具变量 Z 独立于混杂因子）n_samples = 5000
Z = np.random.normal(size=n_samples)  # 工具变量（如广告投放随机种子）X = np.random.uniform(size=(n_samples, 3))  # 协变量（用户特征）T = Z + 0.5*X[:,0] + np.random.normal(scale=0.1)  # 处理变量（实际广告曝光）Y = 0.8*T + 1.2*X[:,1] - 0.4*X[:,2] + np.random.normal()  # 结果（购买金额）

from sklearn.ensemble import RandomForestRegressor

est = RandomForestRegressor(n_estimators=100)
iv_est = DMLIV(
    model_y=est,  # 结果模型
    model_t=est,  # 处理模型
    model_final=est,  # 最终效应模型
    discrete_treatment=False
)
iv_est.fit(Y, T, Z=Z, X=X)

import matplotlib.pyplot as plt

effects = iv_est.effect(X)
plt.errorbar(range(len(X)), effects, yerr=1.96*np.std(effects)/np.sqrt(len(X)))
plt.xlabel("Sample Index")
plt.ylabel("Conditional Average Treatment Effect (CATE)")

当样本量 <5k 时，建议使用 自助法（Bootstrap）计算置信区间：

from econml.bootstrap import BootstrapEstimator

boot_est = BootstrapEstimator(iv_est, n_bootstrap_samples=500)
boot_est.fit(Y, T, Z=Z, X=X)
print(boot_est.interval_(alpha=0.1))  # 输出 90% 置信区间

使用 pywhy 库检查特征间的因果关系是否满足 DAG 假设：

import pywhy.graphs as pg

dag = pg.DAG()
dag.add_edge("Z", "T")  # 工具变量影响处理
dag.add_edge("T", "Y")  # 处理影响结果
dag.add_edge("X1", "Y")  # 协变量影响结果

# 验证是否满足后门准则
assert pg.d_separated(dag, {"T"}, {"Y"}, conditioned_on=["X1", "X2"])

对于超 10 万样本的数据，建议通过 Ray 并行化：

from ray.util.joblib import register_ray
from sklearn.utils import parallel_backend

register_ray()
with parallel_backend('ray'):
    iv_est.fit(Y_large, T_large, Z=Z_large, X=X_large)