因果推断实战指南：从理论到Python实现

1次阅读

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

在日常数据分析中，我们经常遇到这样的场景：发现用户点击广告后购买率提升了 20%，就得出广告有效的结论。但这里可能存在 混淆变量——比如可能是平台同时做了促销活动，真正带动购买的是促销而非广告。这种把相关性当作因果性的错误，在商业决策中可能导致数百万的无效广告投入。

举个真实案例：某电商发现 VIP 用户的留存率比普通用户高，于是大力推广 VIP 会员。但后来发现，VIP 用户本身消费能力更强，即使不办会员也会留存。这就是典型的混淆变量（消费能力）导致的错误归因。

定义每个个体存在两种潜在状态：
$$Y_i(1) \text{（接受干预）}, Y_i(0) \text{（未接受干预）}$$

实际只能观测到其中一种，这就是著名的 因果推断根本问题。我们通常估计 ATE（平均处理效应）：
$$ATE = E[Y(1) – Y(0)]$$

通过有向无环图（DAG）显式建模变量间关系，例如：

graph LR
    A[广告曝光] --> B[购买行为]
    C[用户活跃度] --> A
    C --> B

这里用户活跃度就是混淆变量，必须通过后门调整等方法控制。

ATE：全体用户的平均效应
CATE（条件平均处理效应）：对特定用户群的效应，比如：
$$CATE = E[Y(1)-Y(0)|X=x]$$
在个性化推荐、精准营销中特别有用。

# 类型标注的导入方式
from typing import Tuple, Dict
import numpy as np
import pandas as pd
from sklearn.datasets import make_classification
from dowhy import CausalModel
from econml.dml import LinearDML

# 生成模拟数据
def generate_data(n_samples=1000) -> Tuple[pd.DataFrame, Dict]:
    X, _ = make_classification(n_samples=n_samples, n_features=5, n_informative=3)
    df = pd.DataFrame(X, columns=['age', 'income', 'activity', 'region', 'gender'])
    df['treatment'] = np.random.binomial(1, 0.3 + 0.2*df['activity'])  # 模拟非随机实验
    df['outcome'] = 2*df['treatment'] + 0.5*df['income'] + np.random.normal(0, 1)
    return df, {'confounders': ['activity', 'income']}

df, context = generate_data()

model = CausalModel(
    data=df,
    treatment='treatment',
    outcome='outcome',
    graph="""
    digraph {
        activity -> treatment;
        income -> outcome;
        activity -> outcome;
        treatment -> outcome;
    }
    """
)

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

# 使用 LinearDML 估计异质处理效应
dml_estimate = LinearDML(model_y=LinearRegression(),
                        model_t=LogisticRegression(),
                        discrete_treatment=True)

# 拟合模型
dml_estimate.fit(Y=df['outcome'].values,
    T=df['treatment'].values,
    X=df[['age', 'region', 'gender']],  # 协变量
    W=df[['activity', 'income']]  # 混淆变量
)

# 预测不同用户的 CATE
cate = dml_estimate.effect(df[['age', 'region', 'gender']])

# 安慰剂测试：用随机变量替换真实 treatment
df['placebo'] = np.random.permutation(df['treatment'])
placebo_estimate = model.estimate_effect(
    identified_estimand,
    method_name="backdoor.propensity_score_stratification",
    treatment='placebo'
)
print(f"安慰剂效应: {placebo_estimate.value}")  # 应该接近 0

重叠假设不满足
问题：某些用户群体永远接受 / 不接受干预（如仅向活跃用户发优惠券）
检测：检查倾向得分分布是否重叠
解决：限制分析范围或使用 trimming
时变混淆变量
问题：用户行为随时间变化影响干预分配（如用户变活跃后才被投放广告）
解决：使用边际结构模型 (MSM) 或工具变量
样本选择偏差
问题：分析仅基于留存用户（幸存者偏差）
检测：比较流失 / 留存用户特征
解决：使用逆概率加权(IPW)

当数据量超过单机内存时：

使用 Spark 实现分布式计算：

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression

# 在 Spark 集群上运行双重机器学习
assembler = VectorAssembler(inputCols=['age', 'income', 'activity'],
    outputCol='features'
)

lr = LinearRegression(featuresCol='features', labelCol='outcome')
# 分群体并行拟合模型