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:
- S-learner (Single Model)
- T-learner (Two Models)
- X-learner (Cross-learner)
- R-learner (Robinson Learner)
S-learner: 单一模型
算法
核心思想: 用一个模型同时学习和的关系
步骤:
1. 训练单一模型:
使用任意ML算法(如Random Forest):
model = RandomForestRegressor()
model.fit(np.column_stack([X, D]), Y)2. 预测CATE:
tau_hat = model.predict(np.column_stack([x, [1]])) - \
model.predict(np.column_stack([x, [0]]))优缺点
优势:
- 简单,易实现
- 样本利用率高(用全部数据)
- 正则化效果好
劣势:
- 如果的主效应强,信号可能被淹没
- 不适合处理组和对照组差异大的情况
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. 训练两个模型:
model_0 = RandomForestRegressor().fit(X[D==0], Y[D==0])
model_1 = RandomForestRegressor().fit(X[D==1], Y[D==1])2. 预测CATE:
tau_hat = model_1.predict(x) - model_0.predict(x)优缺点
优势:
- 允许两组有不同的预测模式
- 不受主效应干扰
劣势:
- 如果一组样本很小,该模型可能过拟合
- 两个模型的误差会累积
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实现
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方法:
m_hat = RandomForestRegressor().fit(X, Y).predict(X)
e_hat = LogisticRegression().fit(X, D).predict_proba(X)[:, 1]第2步: 构造残差
第3步: 估计CATE
解优化问题:
使用加权回归:
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实现
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实现:策略评估
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:完整实现
安装
pip install econml完整案例
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的因果革命
异质性因果推断:个性化决策的科学基础!