概述
异质性因果效应(Heterogeneous Treatment Effects, HTE)指的是因果效应在不同个体之间可能存在的差异。与估计平均因果效应(Average Treatment Effect, ATE)不同,HTE 关注的是谁从干预中获益更多,这对于精准医疗、个性化营销等应用至关重要。1
基础定义
潜在结果框架
设:
- :个体 接受干预后的潜在结果
- :个体 未接受干预的潜在结果
- :干预指示变量
关键因果量
| 量 | 定义 | 用途 |
|---|---|---|
| ATE | 总体平均效应 | |
| ATT | 处理组的平均效应 | |
| CATE | 给定特征 的条件效应 | |
| ITE | 个体因果效应 |
核心问题:我们无法直接观察到 ITE,因为每个个体只能处于一种干预状态下。
CATE 的可识别性
识别假设
CATE 的估计需要以下假设:
假设 1:一致性(Consistency)
即实际观察到的结果等于潜在结果的加权平均。
假设 2:互不干涉(No Interference)
个体 的潜在结果不受其他个体干预状态的影响。
假设 3:可忽略性(Ignorability)
给定协变量 ,干预分配与潜在结果条件独立。
假设 4:正值(Positivity)
每个特征的个体都有一定概率接受或不接受干预。
识别公式
在上述假设下:
Meta-Learner 深入分析
S-Learner
方法:训练单个模型联合预测
理论性质:
# S-Learner 的理论分析
class SLearnerAnalysis:
@staticmethod
def bias_analysis():
"""
S-Learner 的偏差分析
如果真实模型是:
Y = f(X) + τ(X)·T + ε
S-Learner 联合模型的最优预测为:
E[Y|X,T] = E[f(X)|X] + E[τ(X)|X]·T + E[ε|X,T]
= f(X) + τ(X)·T
因此uplift估计:
E[Y|X,T=1] - E[Y|X,T=0] = τ(X)
结论:如果模型类足够灵活,S-Learner 是一致的
"""
pass优点:简单,不需要额外的模型
缺点:
- 如果 与 相关较弱,模型可能忽略
- 无法利用倾向得分信息
T-Learner
方法:分别训练两个模型
理论分析:
import numpy as np
def theoretical_analysis_tlearner(n_t, n_c, sigma2):
"""
T-Learner 的理论方差分析
对于叶子节点:
- 干预组估计: \hat{μ}_1 = (1/n_t) Σ Y_i, i∈T=1
- 对照组估计: \hat{μ}_0 = (1/n_c) Σ Y_i, i∈T=0
估计的方差:
Var(\hat{τ}) = Var(\hat{μ}_1) + Var(\hat{μ}_0)
= σ²/n_t + σ²/n_c
= σ² · (1/n_t + 1/n_c)
"""
variance = sigma2 * (1/n_t + 1/n_c)
std_error = np.sqrt(variance)
return variance, std_error优点:每个模型专注于特定干预水平
缺点:
- 当样本不平衡时,某一模型可能欠拟合
- 无法利用倾向得分
X-Learner
改进:解决样本不平衡问题
class XLearnerAdvanced:
@staticmethod
def detailed_steps(X, T, Y, propensity_model=None):
"""
X-Learner 的详细步骤
Step 1: 训练响应曲面
- \hat{μ}_1(x) ≈ E[Y|T=1,X=x]
- \hat{μ}_0(x) ≈ E[Y|T=0,X=x]
Step 2: 计算伪结果(Imputed Potential Outcomes)
- 对于干预组: D₁ = Y - \hat{μ}_0(X) (反事实对照)
- 对于对照组: D₀ = \hat{μ}_1(X) - Y (反事实干预)
Step 3: 估计个体治疗效应
- \hat{τ}_1(x) ≈ E[D₁|X=x]
- \hat{τ}_0(x) ≈ E[D₀|X=x]
Step 4: 组合两个ITE估计
- \hat{τ}(x) = g(x)·\hat{τ}_1(x) + (1-g(x))·\hat{τ}_0(x)
其中 g(x) 是倾向得分相关的权重
"""
passDR-Learner(双重稳健估计)
核心思想:结合结果模型和倾向得分模型,只要其中之一正确即一致。2
class DRLearner:
def __init__(self, outcome_model, propensity_model):
self.outcome_model = outcome_model
self.propensity_model = propensity_model
def fit(self, X, T, Y):
"""
DR-Learner 的拟合过程
对于每个交叉验证折:
1. 在训练集上训练 outcome_model 和 propensity_model
2. 在验证集上计算 DR 量
"""
pass
def predict_cate(self, X_test):
"""
DR 估计量:
\hat{τ}_DR(x) = \hat{m}_1(x) - \hat{m}_0(x)
+ T·(Y - \hat{m}_1(x))/\hat{e}(x)
- (1-T)·(Y - \hat{m}_0(x))/(1-\hat{e}(x))
其中 \hat{e}(x) = P(T=1|X=x) 是倾向得分
"""
pass数学推导:
DR 估计量的一致性:
即使倾向得分模型错误,只要结果模型正确,估计仍然一致。
不确定性量化
置信区间构建
由于每个个体只能观察到一种潜在结果,CATE 的估计本质上存在不确定性。
方法 1:Bootstrap
import numpy as np
from sklearn.utils import resample
def bootstrap_confidence_interval(X, T, Y, model, n_bootstrap=100, alpha=0.05):
"""
Bootstrap 方法构建 CATE 估计的置信区间
Args:
X: 特征矩阵
T: 干预指示
Y: 结果
model: CATE 估计模型
n_bootstrap: Bootstrap 次数
alpha: 显著性水平
"""
n = len(Y)
tau_estimates = []
for _ in range(n_bootstrap):
# Bootstrap 采样
idx = resample(range(n))
X_boot = X[idx]
T_boot = T[idx]
Y_boot = Y[idx]
# 训练模型并预测
model.fit(X_boot, T_boot, Y_boot)
tau_hat = model.predict_cate(X)
tau_estimates.append(tau_hat)
tau_estimates = np.array(tau_estimates)
# 计算置信区间
lower = np.percentile(tau_estimates, 100 * alpha / 2, axis=0)
upper = np.percentile(tau_estimates, 100 * (1 - alpha / 2), axis=0)
return lower, upper方法 2:基于方差估计的区间
def variance_based_ci(X, T, Y, model):
"""
基于理论方差的置信区间
对于 CATE 估计:
Var(\hat{τ}(x)) ≈ Var(Y|T=1,X) / P(T=1|X) + Var(Y|T=0,X) / P(T=0|X)
渐近 95% 置信区间:
\hat{τ}(x) ± 1.96 · √(Var(\hat{τ}(x)))
"""
# 估计条件方差
var_t1 = model.estimate_variance(X, T, Y, treatment=1)
var_t0 = model.estimate_variance(X, T, Y, treatment=0)
# 估计倾向得分
prop = model.estimate_propensity(X)
# 计算方差
var_cate = var_t1 / (prop + 1e-5) + var_t0 / (1 - prop + 1e-5)
# 置信区间
tau_hat = model.predict_cate(X)
lower = tau_hat - 1.96 * np.sqrt(var_cate)
upper = tau_hat + 1.96 * np.sqrt(var_cate)
return lower, upper贝叶斯方法
import torch
import torch.nn as nn
from torch.distributions import Normal
class BayesianCATE(nn.Module):
"""
贝叶斯 CATE 估计器
使用变分推断近似后验分布:
q(θ) ≈ p(θ|D)
从而获得预测的不确定性:
P(τ(x)|D) ≈ ∫ P(τ(x)|θ) q(θ) dθ
"""
def __init__(self, input_dim, hidden_dim=64):
super().__init__()
# 共享特征提取
self.feature_net = nn.Sequential(
nn.Linear(input_dim, hidden_dim),
nn.ReLU(),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU()
)
# 干预组预测(均值和方差)
self.outcome_mean_t1 = nn.Linear(hidden_dim, 1)
self.outcome_log_var_t1 = nn.Linear(hidden_dim, 1)
# 对照组预测
self.outcome_mean_t0 = nn.Linear(hidden_dim, 1)
self.outcome_log_var_t0 = nn.Linear(hidden_dim, 1)
def forward(self, x, t):
features = self.feature_net(x)
if t == 1:
mean = self.outcome_mean_t1(features)
log_var = self.outcome_log_var_t1(features)
else:
mean = self.outcome_mean_t0(features)
log_var = self.outcome_log_var_t0(features)
return mean, log_var
def predict_cate_with_uncertainty(self, x, n_samples=100):
"""
蒙特卡洛dropout估计CATE的后验分布
"""
self.train() # 启用 dropout
cate_samples = []
for _ in range(n_samples):
mean_t1, _ = self.forward(x, t=1)
mean_t0, _ = self.forward(x, t=0)
cate_samples.append(mean_t1 - mean_t0)
cate_samples = torch.cat(cate_samples, dim=0)
# 后验均值和标准差
cate_mean = cate_samples.mean(dim=0)
cate_std = cate_samples.std(dim=0)
return cate_mean, cate_stdCATE 的统计性质
一致性
当样本量 时:
渐近正态性
其中 是 CATE 估计的渐近方差。
收敛速度
对于基于树的 CATE 估计:
其中 是潜在函数的平滑度, 是特征维度。
实践注意事项
1. 样本量要求
def sample_size_calculator(delta, sigma2, power=0.8, alpha=0.05):
"""
检测 CATE 的最小样本量计算
H₀: τ(x) = 0
H₁: |τ(x)| > δ
所需样本量:
n ≈ 2 · σ² · (Z_{1-α/2} + Z_{power})² / δ²
"""
from scipy.stats import norm
z_alpha = norm.ppf(1 - alpha/2)
z_power = norm.ppf(power)
n = 2 * sigma2 * (z_alpha + z_power)**2 / delta**2
return int(np.ceil(n))
# 示例:检测 5% 的 uplift,标准差 0.3
n_required = sample_size_calculator(delta=0.05, sigma2=0.09)
print(f"每组至少需要 {n_required} 个样本")2. 倾向得分重叠区域
def check_overlap(X, T, propensity_model=None):
"""
检查倾向得分的重叠情况
重叠假设(Positivity)要求:
0 < P(T=1|X) < 1 对所有 X
如果某些区域只有干预组或对照组,需要外推
"""
from sklearn.linear_model import LogisticRegression
if propensity_model is None:
propensity_model = LogisticRegression()
propensity_model.fit(X, T)
prop = propensity_model.predict_proba(X)[:, 1]
# 检查重叠
overlap = (prop > 0.1) & (prop < 0.9)
overlap_rate = overlap.mean()
print(f"重叠区域比例: {overlap_rate:.2%}")
print(f"倾向得分范围: [{prop.min():.3f}, {prop.max():.3f}]")
# 识别外推区域
if prop.max() < 0.99 and prop.min() > 0.01:
print("✓ 倾向得分重叠良好")
else:
print("⚠ 存在外推区域,需谨慎解释")
return overlap, prop3. 模型选择
from sklearn.model_selection import cross_val_score
def compare_meta_learners(X, T, Y, models):
"""
比较不同 meta-learner 的性能
"""
results = {}
for name, model in models.items():
# 使用 5 折交叉验证
scores = cross_val_score(model, X, T, Y, cv=5, scoring='uplift_auc')
results[name] = {
'mean': scores.mean(),
'std': scores.std(),
'scores': scores
}
return pd.DataFrame(results).T4. 避免常见陷阱
| 陷阱 | 描述 | 解决方案 |
|---|---|---|
| 过拟合 | 在训练数据上表现好但泛化差 | 使用交叉验证 |
| 选择偏差 | 非随机分组导致估计偏差 | 使用倾向得分加权 |
| 外推风险 | 对重叠区域外的预测不可靠 | 检查倾向得分分布 |
| 多重比较 | 检验太多导致假阳性 | 调整显著性水平 |
与 Uplift Modeling 的关系
HTE 估计和 Uplift Modeling 本质上是同一问题的不同视角:
- HTE:从因果推断理论出发,强调识别条件和统计性质
- Uplift Modeling:从应用角度出发,强调排序能力和业务指标
两种方法的结合:
- 使用 HTE 的识别理论保证因果性
- 使用 Uplift 的评估指标(Qini、AUUC)评估实际效果