Skip to content

13.4 异质性处理效应与Meta-learners

"Personalized treatment rules can substantially improve outcomes compared to one-size-fits-all policies.""与一刀切的政策相比,个性化处理规则可以大幅改善结果。"— Susan Athey & Stefan Wager

Susan Athey & Stefan Wager:从平均效应到个性化因果推断

难度方法应用


本节目标

  • 深刻理解CATE(条件平均处理效应)的重要性
  • 掌握S-learner, T-learner, X-learner, R-learner的原理
  • 深入理解Causal Forest(扩展Module 12)
  • 学习最优政策学习(Policy Learning)
  • 使用EconML完整实现所有Meta-learners
  • 应用于真实个性化决策场景

引言:平均效应的局限

ATE的问题

传统因果推断: 估计平均处理效应(ATE)

例子: 教育培训项目

传统结论: "培训平均提高收入$5,000"

政策:给所有人培训

异质性的价值

现实: 处理效应因人而异!

假设真实情况:

  • 年轻人(20-30岁):
  • 中年人(30-50岁):
  • 老年人(50+岁): (负效应!)

更好的政策:

  • 重点投资年轻人
  • 节约资源,避免伤害

社会福利提升:


条件平均处理效应(CATE)

定义

CATE (Conditional Average Treatment Effect):

解释: 在给定协变量的条件下,处理的平均因果效应。

关键差异:

概念定义异质性应用
ATE忽略平均政策
ATT部分已处理者
CATE完全个性化
ITE个体不可识别

重要性质:

1. 平均性:

2. 政策价值:

其中是政策函数。

3. 最优政策:

只给那些的人处理!


Meta-learners框架

核心思想

Meta-learner: 用机器学习算法作为"基学习器"估计CATE

通用框架:

第1步: 选择基学习器(如Random Forest, XGBoost)
第2步: 用特定方式训练模型
第3步: 预测CATE: τ̂(x)

4大Meta-learners:

  1. S-learner (Single Model)
  2. T-learner (Two Models)
  3. X-learner (Cross-learner)
  4. R-learner (Robinson Learner)

S-learner: 单一模型

算法

核心思想: 用一个模型同时学习的关系

步骤:

1. 训练单一模型:

使用任意ML算法(如Random Forest):

python
model = RandomForestRegressor()
model.fit(np.column_stack([X, D]), Y)

2. 预测CATE:

python
tau_hat = model.predict(np.column_stack([x, [1]])) - \
          model.predict(np.column_stack([x, [0]]))

优缺点

优势:

  • 简单,易实现
  • 样本利用率高(用全部数据)
  • 正则化效果好

劣势:

  • 如果的主效应强,信号可能被淹没
  • 不适合处理组和对照组差异大的情况

Python实现

python
import numpy as np
from sklearn.ensemble import RandomForestRegressor

class SLearner:
    """S-learner元学习器"""

    def __init__(self, model=None):
        self.model = model or RandomForestRegressor(n_estimators=100, random_state=42)

    def fit(self, X, D, Y):
        """
        训练S-learner

        参数:
        ------
        X: (N, K) 协变量
        D: (N,) 处理指示器
        Y: (N,) 结果变量
        """
        # 合并X和D
        X_augmented = np.column_stack([X, D])
        self.model.fit(X_augmented, Y)
        self.n_features = X.shape[1]
        return self

    def predict(self, X):
        """预测CATE"""
        N = X.shape[0]

        # 预测Y(1)
        X_1 = np.column_stack([X, np.ones(N)])
        Y_1_pred = self.model.predict(X_1)

        # 预测Y(0)
        X_0 = np.column_stack([X, np.zeros(N)])
        Y_0_pred = self.model.predict(X_0)

        # CATE
        tau = Y_1_pred - Y_0_pred
        return tau

# 使用示例
slearner = SLearner()
slearner.fit(X, D, Y)
tau_s = slearner.predict(X_test)

T-learner: 两个模型

算法

核心思想: 分别对处理组和对照组训练独立模型

步骤:

1. 训练两个模型:

python
model_0 = RandomForestRegressor().fit(X[D==0], Y[D==0])
model_1 = RandomForestRegressor().fit(X[D==1], Y[D==1])

2. 预测CATE:

python
tau_hat = model_1.predict(x) - model_0.predict(x)

优缺点

优势:

  • 允许两组有不同的预测模式
  • 不受主效应干扰

劣势:

  • 如果一组样本很小,该模型可能过拟合
  • 两个模型的误差会累积

Python实现

python
class TLearner:
    """T-learner元学习器"""

    def __init__(self, model_0=None, model_1=None):
        self.model_0 = model_0 or RandomForestRegressor(n_estimators=100, random_state=42)
        self.model_1 = model_1 or RandomForestRegressor(n_estimators=100, random_state=43)

    def fit(self, X, D, Y):
        """训练两个独立模型"""
        # 对照组模型
        self.model_0.fit(X[D==0], Y[D==0])

        # 处理组模型
        self.model_1.fit(X[D==1], Y[D==1])

        return self

    def predict(self, X):
        """预测CATE"""
        mu_0 = self.model_0.predict(X)
        mu_1 = self.model_1.predict(X)
        tau = mu_1 - mu_0
        return tau

# 使用
tlearner = TLearner()
tlearner.fit(X, D, Y)
tau_t = tlearner.predict(X_test)

X-learner: 交叉学习器

算法 (Künzel et al. 2019)

核心思想: 利用已训练的模型构造"伪结果",再训练CATE模型

步骤:

第1阶段: 训练基础模型(同T-learner)

第2阶段: 构造"伪处理效应"

对于对照组:

对于处理组:

直觉:

  • : "如果这个对照个体接受处理,效应是多少?"
  • : "如果这个处理个体不接受处理,效应是多少?"

第3阶段: 训练CATE模型

第4阶段: 加权组合

其中权重函数:

是倾向得分。

直觉:

  • 如果高(处理概率大),多依赖(来自处理组)
  • 如果低(对照概率大),多依赖(来自对照组)

优势

Künzel et al. (2019)定理:

当一组样本量远大于另一组时,X-learner的MSE显著低于T-learner!

原因: 利用大样本组的信息帮助小样本组估计。

Python实现

python
from sklearn.linear_model import LogisticRegression

class XLearner:
    """X-learner元学习器"""

    def __init__(self, model_0=None, model_1=None, model_tau_0=None, model_tau_1=None):
        self.model_0 = model_0 or RandomForestRegressor(n_estimators=100, random_state=42)
        self.model_1 = model_1 or RandomForestRegressor(n_estimators=100, random_state=43)
        self.model_tau_0 = model_tau_0 or RandomForestRegressor(n_estimators=50, random_state=44)
        self.model_tau_1 = model_tau_1 or RandomForestRegressor(n_estimators=50, random_state=45)
        self.ps_model = LogisticRegression(max_iter=1000)

    def fit(self, X, D, Y):
        """训练X-learner"""
        # 第1阶段:基础模型
        self.model_0.fit(X[D==0], Y[D==0])
        self.model_1.fit(X[D==1], Y[D==1])

        # 第2阶段:构造伪结果
        # 对照组的伪效应
        X_0 = X[D==0]
        Y_0 = Y[D==0]
        mu_1_on_0 = self.model_1.predict(X_0)
        tau_tilde_0 = mu_1_on_0 - Y_0

        # 处理组的伪效应
        X_1 = X[D==1]
        Y_1 = Y[D==1]
        mu_0_on_1 = self.model_0.predict(X_1)
        tau_tilde_1 = Y_1 - mu_0_on_1

        # 第3阶段:训练CATE模型
        self.model_tau_0.fit(X_0, tau_tilde_0)
        self.model_tau_1.fit(X_1, tau_tilde_1)

        # 估计倾向得分
        self.ps_model.fit(X, D)

        return self

    def predict(self, X):
        """预测CATE"""
        tau_0 = self.model_tau_0.predict(X)
        tau_1 = self.model_tau_1.predict(X)

        # 倾向得分
        e = self.ps_model.predict_proba(X)[:, 1]

        # 加权组合
        g = 1 - e  # 权重
        tau = g * tau_0 + (1 - g) * tau_1

        return tau

# 使用
xlearner = XLearner()
xlearner.fit(X, D, Y)
tau_x = xlearner.predict(X_test)

R-learner: Robinson分解

算法 (Nie & Wager 2021)

核心思想: 先去除混淆,再估计CATE

Robinson分解 (1988):

其中:

  • : 无条件结果函数
  • : 倾向得分
  • : CATE

步骤:

第1步: 估计

使用任意ML方法:

python
m_hat = RandomForestRegressor().fit(X, Y).predict(X)
e_hat = LogisticRegression().fit(X, D).predict_proba(X)[:, 1]

第2步: 构造残差

第3步: 估计CATE

解优化问题:

使用加权回归:

python
from sklearn.ensemble import GradientBoostingRegressor

tau_model = GradientBoostingRegressor()
tau_model.fit(X, Y_tilde, sample_weight=np.abs(D_tilde))
tau_hat = tau_model.predict(X_test)

理论性质

Nie & Wager (2021)定理:

R-learner在温和条件下达到半参数效率界(Semiparametric Efficiency Bound)!

含义: 在所有估计量中,R-learner的渐近方差最小。

Python实现

python
class RLearner:
    """R-learner元学习器"""

    def __init__(self, model_m=None, model_e=None, model_tau=None):
        self.model_m = model_m or RandomForestRegressor(n_estimators=100, random_state=42)
        self.model_e = model_e or LogisticRegression(max_iter=1000)
        self.model_tau = model_tau or GradientBoostingRegressor(n_estimators=100, random_state=43)

    def fit(self, X, D, Y):
        """训练R-learner"""
        # 第1步:估计m(X)和e(X)
        self.model_m.fit(X, Y)
        m_hat = self.model_m.predict(X)

        self.model_e.fit(X, D)
        e_hat = self.model_e.predict_proba(X)[:, 1]

        # 第2步:构造残差
        Y_tilde = Y - m_hat
        D_tilde = D - e_hat

        # 第3步:加权回归估计τ(X)
        # 权重:避免除以零
        weights = np.abs(D_tilde) + 1e-6

        self.model_tau.fit(X, Y_tilde / (D_tilde + 1e-6), sample_weight=weights)

        return self

    def predict(self, X):
        """预测CATE"""
        tau = self.model_tau.predict(X)
        return tau

# 使用
rlearner = RLearner()
rlearner.fit(X, D, Y)
tau_r = rlearner.predict(X_test)

Causal Forest深入(回顾Module 12)

核心创新

Wager & Athey (2018) 的Causal Forest是特殊的Meta-learner:

关键特点:

1. 诚实分割(Honest Splitting)

  • 样本分割:一半建树,一半估计
  • 避免过拟合

2. 局部随机化

  • 每个叶子内:处理和对照"近似RCT"

3. 自适应

  • 自动寻找异质性最大的区域

与Meta-learners对比

方法理论保证推断速度解释性
S-learner
T-learner
X-learner(有界)
R-learner(效率)
Causal Forest(渐近正态)置信区间

何时用Causal Forest?

优先CF:

  • 需要置信区间和统计推断
  • 样本量大()
  • 异质性复杂(非线性)

优先Meta-learners:

  • 样本量中等
  • 需要快速迭代
  • 想用其他ML算法(XGBoost等)

最优政策学习

从CATE到政策

目标: 设计政策最大化社会福利

社会福利函数:

其中是处理成本。

最优政策:

解释: 只给那些收益超过成本的人处理!

策略价值估计

问题: 如何评估政策的价值?

IPW估计量 (Dudík et al. 2011):

DR估计量:

Python实现:策略评估

python
def evaluate_policy(pi, X, D, Y, e_hat, mu_1_hat):
    """
    评估政策价值

    参数:
    ------
    pi: (N,) 策略(0/1)
    X: (N, K) 协变量
    D: (N,) 实际处理
    Y: (N,) 结果
    e_hat: (N,) 倾向得分估计
    mu_1_hat: (N,) E[Y|D=1,X]估计

    返回:
    ------
    value: 策略价值
    """
    N = len(Y)

    # DR估计
    term1 = pi * D * (Y - mu_1_hat) / (e_hat + 1e-6)
    term2 = pi * mu_1_hat

    value = np.mean(term1 + term2)

    return value

# 比较不同政策
def compare_policies(tau_hat, X, D, Y, e_hat, mu_1_hat, cost=0):
    """比较多种策略"""

    # 政策1:给所有人
    pi_all = np.ones(len(X))
    value_all = evaluate_policy(pi_all, X, D, Y, e_hat, mu_1_hat)

    # 政策2:给τ(x)>0的人
    pi_positive = (tau_hat > 0).astype(int)
    value_positive = evaluate_policy(pi_positive, X, D, Y, e_hat, mu_1_hat)

    # 政策3:给τ(x)>cost的人
    pi_cost = (tau_hat > cost).astype(int)
    value_cost = evaluate_policy(pi_cost, X, D, Y, e_hat, mu_1_hat)

    results = pd.DataFrame({
        '政策': ['全部处理', 'τ>0', f'τ>{cost}'],
        '价值': [value_all, value_positive, value_cost],
        '处理比例': [pi_all.mean(), pi_positive.mean(), pi_cost.mean()]
    })

    return results

️ 使用EconML:完整实现

安装

bash
pip install econml

完整案例

python
import numpy as np
import pandas as pd
from econml.metalearners import SLearner, TLearner, XLearner
from econml.dml import CausalForestDML, LinearDML
from econml.cate_interpreter import SingleTreeCateInterpreter
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
import matplotlib.pyplot as plt
import seaborn as sns

plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']
plt.rcParams['axes.unicode_minus'] = False

# =================================================================
# 第1步:生成异质性数据
# =================================================================

np.random.seed(42)
N = 5000

# 协变量
X = pd.DataFrame({
    'age': np.random.uniform(20, 65, N),
    'education': np.random.uniform(8, 20, N),
    'experience': np.random.uniform(0, 30, N),
    'female': np.random.binomial(1, 0.5, N),
    'urban': np.random.binomial(1, 0.6, N)
})

# 倾向得分(混淆)
logit_e = -1 + 0.05*X['age'] + 0.1*X['education'] - 0.5*X['female']
e_true = 1 / (1 + np.exp(-logit_e))
D = (np.random.rand(N) < e_true).astype(int)

# 异质性处理效应
tau_true = 1000 + 500*X['age']/40 - 200*X['age']**2/1600 + 300*X['education']/10

# 结果变量
Y0 = 10000 + 500*X['age']/40 + 1000*X['education']/10 + 1000*X['experience']/10 + \
     np.random.normal(0, 2000, N)
Y1 = Y0 + tau_true
Y = D * Y1 + (1-D) * Y0

print(f"样本量: {N}")
print(f"处理组: {D.sum()} ({D.mean()*100:.1f}%)")
print(f"真实ATE: ${tau_true.mean():.0f}")
print(f"真实τ范围: [${tau_true.min():.0f}, ${tau_true.max():.0f}]")

# =================================================================
# 第2步:估计CATE - 所有Meta-learners
# =================================================================

X_arr = X.values

# S-Learner
slearner = SLearner(overall_model=RandomForestRegressor(n_estimators=100, random_state=42))
slearner.fit(Y, D, X=X_arr)
tau_s = slearner.effect(X_arr)

# T-Learner
tlearner = TLearner(models=[RandomForestRegressor(n_estimators=100, random_state=42),
                            RandomForestRegressor(n_estimators=100, random_state=43)])
tlearner.fit(Y, D, X=X_arr)
tau_t = tlearner.effect(X_arr)

# X-Learner
xlearner = XLearner(models=[RandomForestRegressor(n_estimators=100, random_state=42),
                            RandomForestRegressor(n_estimators=100, random_state=43)],
                    cate_models=[RandomForestRegressor(n_estimators=50, random_state=44),
                                 RandomForestRegressor(n_estimators=50, random_state=45)])
xlearner.fit(Y, D, X=X_arr)
tau_x = xlearner.effect(X_arr)

# Causal Forest (DML版本)
cf = CausalForestDML(model_y=RandomForestRegressor(n_estimators=100, random_state=46),
                     model_t=RandomForestRegressor(n_estimators=100, random_state=47),
                     n_estimators=100,
                     random_state=48)
cf.fit(Y, D, X=X_arr)
tau_cf = cf.effect(X_arr)

# =================================================================
# 第3步:性能评估
# =================================================================

from sklearn.metrics import mean_squared_error, r2_score

results = pd.DataFrame({
    'Method': ['S-Learner', 'T-Learner', 'X-Learner', 'Causal Forest'],
    'RMSE': [
        np.sqrt(mean_squared_error(tau_true, tau_s)),
        np.sqrt(mean_squared_error(tau_true, tau_t)),
        np.sqrt(mean_squared_error(tau_true, tau_x)),
        np.sqrt(mean_squared_error(tau_true, tau_cf))
    ],
    'R²': [
        r2_score(tau_true, tau_s),
        r2_score(tau_true, tau_t),
        r2_score(tau_true, tau_x),
        r2_score(tau_true, tau_cf)
    ],
    'ATE_Error': [
        abs(tau_s.mean() - tau_true.mean()),
        abs(tau_t.mean() - tau_true.mean()),
        abs(tau_x.mean() - tau_true.mean()),
        abs(tau_cf.mean() - tau_true.mean())
    ]
})

print("\n性能比较:")
print(results.to_string(index=False))

# =================================================================
# 第4步:可视化异质性
# =================================================================

fig, axes = plt.subplots(2, 3, figsize=(18, 12))

methods = [
    ('真实CATE', tau_true),
    ('S-Learner', tau_s),
    ('T-Learner', tau_t),
    ('X-Learner', tau_x),
    ('Causal Forest', tau_cf)
]

for idx, (name, tau) in enumerate(methods):
    row = idx // 3
    col = idx % 3
    ax = axes[row, col]

    scatter = ax.scatter(X['age'], tau, c=X['education'],
                        cmap='viridis', alpha=0.5, s=10)
    ax.set_xlabel('年龄', fontsize=11)
    ax.set_ylabel('CATE', fontsize=11)
    ax.set_title(f'{name}\nATE = ${tau.mean():.0f}', fontsize=12, fontweight='bold')
    plt.colorbar(scatter, ax=ax, label='教育年限')

# 方法比较
ax = axes[1, 2]
for name, tau in methods[1:]:
    ax.scatter(tau_true, tau, alpha=0.3, s=5, label=name)
ax.plot([tau_true.min(), tau_true.max()],
        [tau_true.min(), tau_true.max()],
        'k--', lw=2, label='完美预测')
ax.set_xlabel('真实CATE', fontsize=11)
ax.set_ylabel('估计CATE', fontsize=11)
ax.set_title('方法比较:真实vs估计', fontsize=12, fontweight='bold')
ax.legend(fontsize=9)

plt.tight_layout()
plt.show()

# =================================================================
# 第5步:策略学习
# =================================================================

# 最优策略(基于真实CATE)
pi_optimal = (tau_true > 0).astype(int)

# 各方法的策略
pi_s = (tau_s > 0).astype(int)
pi_t = (tau_t > 0).astype(int)
pi_x = (tau_x > 0).astype(int)
pi_cf = (tau_cf > 0).astype(int)

# 策略价值
policy_results = pd.DataFrame({
    '策略': ['全部处理', '最优', 'S-Learner', 'T-Learner', 'X-Learner', 'Causal Forest'],
    '平均收益': [
        tau_true.mean(),
        tau_true[pi_optimal==1].mean() if pi_optimal.sum() > 0 else 0,
        tau_true[pi_s==1].mean() if pi_s.sum() > 0 else 0,
        tau_true[pi_t==1].mean() if pi_t.sum() > 0 else 0,
        tau_true[pi_x==1].mean() if pi_x.sum() > 0 else 0,
        tau_true[pi_cf==1].mean() if pi_cf.sum() > 0 else 0
    ],
    '处理比例': [
        1.0,
        pi_optimal.mean(),
        pi_s.mean(),
        pi_t.mean(),
        pi_x.mean(),
        pi_cf.mean()
    ]
})

policy_results['总收益'] = policy_results['平均收益'] * policy_results['处理比例'] * N

print("\n策略评估:")
print(policy_results.to_string(index=False))

# =================================================================
# 第6步:CATE解释(可选)
# =================================================================

# 用单棵树解释Causal Forest的异质性
intrp = SingleTreeCateInterpreter(include_model_uncertainty=False, max_depth=3)
intrp.interpret(cf, X_arr)
intrp.plot(feature_names=X.columns, fontsize=10)
plt.title('Causal Forest异质性解释树', fontsize=14, fontweight='bold')
plt.tight_layout()
plt.show()

print("\n分析完成!")

️ 常见陷阱

陷阱1:混淆ATE和CATE

错误: 报告时忘记它是条件效应

正确: 强调这是"对于特征的个体"的效应

陷阱2:过度拟合

问题: CATE模型可能过拟合噪声

缓解:

  • 交叉验证
  • 正则化
  • 样本分割(诚实估计)

陷阱3:忽略不确定性

问题: CATE估计有不确定性!

解决:

  • Causal Forest提供置信区间
  • Bootstrap估计标准误

陷阱4:策略评估偏差

问题: 用训练数据评估策略 → 过于乐观

正确:

  • Hold-out集评估
  • 交叉拟合

本节小结

核心要点

1. CATE的价值: 从平均效应到个性化因果推断

2. 四大Meta-learners:

  • S-learner: 简单,但信号可能被淹没
  • T-learner: 灵活,但误差累积
  • X-learner: 样本不平衡时最优
  • R-learner: 理论最优(半参数效率)

3. Causal Forest: 唯一提供统计推断的方法

4. 政策学习:

关键公式

CATE:

X-learner:

R-learner:

最优政策价值:

Python工具

  • econml: SLearner, TLearner, XLearner, CausalForestDML
  • causalml: BaseSLearner, BaseTLearner, BaseXLearner
  • sklearn: 基学习器

下一节: 13.5 因果图与结构因果模型 - Judea Pearl的因果革命


异质性因果推断:个性化决策的科学基础!

基于 MIT 许可证发布。内容版权归作者所有。