概述

Uplift Modeling 的方法可以分为四大类:1

  1. Meta-Learner:将 CATE 估计问题转化为标准的监督学习问题
  2. 直接建模法:直接建模 uplift 的分布或目标函数
  3. 基于树的方法:专门设计用于捕捉异质性因果效应的树结构
  4. 深度学习方法:利用神经网络强大的表示学习能力

Meta-Learner

Meta-Learner 的核心思想是将 CATE 估计问题分解为多个标准监督学习子问题,通过组合子模型的预测来估计 uplift。

S-Learner(单模型方法)

原理:使用单个模型,将干预变量 作为特征之一:

其中 是联合预测模型。

Python 实现

from sklearn.ensemble import GradientBoostingClassifier
import numpy as np
 
class SLearner:
    def __init__(self, base_model=None):
        self.base_model = base_model or GradientBoostingClassifier()
    
    def fit(self, X, treatment, y):
        # 将干预作为特征加入
        X_with_t = np.column_stack([X, treatment])
        self.model = self.base_model.__class__(**self.base_model.get_params())
        self.model.fit(X_with_t, y)
    
    def predict_uplift(self, X):
        # 预测干预和不干预两种情况
        X_t1 = np.column_stack([X, np.ones(len(X))])
        X_t0 = np.column_stack([X, np.zeros(len(X))])
        
        p_t1 = self.model.predict_proba(X_t1)[:, 1]
        p_t0 = self.model.predict_proba(X_t0)[:, 1]
        
        return p_t1 - p_t0

优点:简单直接,可以使用任意回归/分类模型
缺点

  • 的基数较高时,每个干预水平需要单独估计
  • 模型可能过度关注预测 而非

T-Learner(双模型方法)

原理:分别训练干预组和对照组的模型:

Python 实现

class TLearner:
    def __init__(self, model=None):
        self.model_1 = (model or GradientBoostingClassifier()).__class__(
            **model.get_params() if model else {}
        )
        self.model_0 = (model or GradientBoostingClassifier()).__class__(
            **model.get_params() if model else {}
        )
    
    def fit(self, X, treatment, y):
        mask_t1 = treatment == 1
        self.model_1.fit(X[mask_t1], y[mask_t1])
        self.model_0.fit(X[~mask_t1], y[~mask_t1])
    
    def predict_uplift(self, X):
        p_t1 = self.model_1.predict_proba(X)[:, 1]
        p_t0 = self.model_0.predict_proba(X)[:, 1]
        return p_t1 - p_t0

优点:模型分工明确,每个模型专注于一个干预水平
缺点:如果某一组的样本量很少,预测效果会很差

X-Learner

原理:T-Learner 的改进版本,通过反事实估计来提高精度:2

步骤

  1. 与 T-Learner 一样,分别训练
  2. 计算伪结果(pseudo-outcomes):
    • 对照组用户:,其中
    • 干预组用户:
  3. 用这些伪结果训练倾向得分加权的 CATE 模型

伪代码

class XLearner:
    def __init__(self, model=None):
        self.model_1 = None
        self.model_0 = None
        self.tau_model = None
    
    def fit(self, X, treatment, y, propensity_model=None):
        # Step 1: 训练基础模型
        mask_t1 = treatment == 1
        mask_t0 = treatment == 0
        
        self.model_1 = self._train_model(X[mask_t1], y[mask_t1])
        self.model_0 = self._train_model(X[mask_t0], y[mask_t0])
        
        # Step 2: 计算伪结果
        tau_hat_t0 = self.model_1.predict(X[mask_t0]) - y[mask_t0]
        tau_hat_t1 = y[mask_t1] - self.model_0.predict(X[mask_t1])
        
        # Step 3: 训练 CATE 模型
        X_cate = np.vstack([X[mask_t0], X[mask_t1]])
        tau_pseudo = np.concatenate([tau_hat_t0, tau_hat_t1])
        
        self.tau_model = self._train_model(X_cate, tau_pseudo)
    
    def predict_uplift(self, X):
        return self.tau_model.predict(X)

CATE-Learner(DR-Learner)

原理:使用双重稳健估计(Doubly Robust)结合倾向得分:3

其中 是倾向得分,

优点:只要倾向得分模型或结果模型之一正确,DR-Learner 就是一致的
缺点:需要估计倾向得分,在随机实验中可以直接用样本比例


直接建模法

Class Transformation

Jaskowski & Jaroszewicz (2012) 提出的方法,将 uplift 估计问题转化为标准二分类问题。4

核心观察

转换目标

则:

从而:

Python 实现

class ClassTransformation:
    def __init__(self, model=None):
        self.model = model or GradientBoostingClassifier()
    
    def fit(self, X, treatment, y):
        # 转换标签
        Z = y * treatment - (1 - y) * (1 - treatment)
        self.model.fit(X, Z)
    
    def predict_uplift(self, X):
        prob = self.model.predict_proba(X)[:, 1]
        return 2 * prob - 1

关键假设:干预组和对照组的样本量相等
注意事项:如果样本不平衡,需要进行加权调整


基于树的方法

树结构天然适合捕捉异质性效应,因为它们通过递归分裂来划分特征空间,每个叶子节点代表一个具有相似效应的群体。

Uplift Tree(Causal Tree 的前身)

分裂准则:最大化干预组和对照组之间的响应率差异

其中 是两组在父节点的样本量, 是左右子节点的平均结果。

Causal Tree(Athey & Imbens)

核心创新:为每个叶子节点估计 CATE 并计算诚实估计(Honest Estimation):1

  1. 将数据随机分为两部分:一部分用于树构建,一部分用于叶子节点估计
  2. 叶子节点的 CATE 估计:
  1. 使用剪枝集来选择最优的树复杂度

Python 实现(简化版)

from sklearn.tree import DecisionTreeClassifier
import numpy as np
 
class CausalTree:
    def __init__(self, max_depth=5, min_samples_leaf=100):
        self.max_depth = max_depth
        self.min_samples_leaf = min_samples_leaf
        self.tree = None
    
    def _causal_split_criterion(self, X, y, treatment, feature_idx, threshold):
        """计算分裂的信息增益"""
        left_mask = X[:, feature_idx] <= threshold
        right_mask = ~left_mask
        
        # 计算左右子节点的 CATE
        def compute_cate(mask):
            if mask.sum() == 0:
                return 0
            t_mask = mask & (treatment == 1)
            c_mask = mask & (treatment == 0)
            if t_mask.sum() == 0 or c_mask.sum() == 0:
                return 0
            return (y[t_mask].sum() / t_mask.sum() - 
                    y[c_mask].sum() / c_mask.sum())
        
        n_left = left_mask.sum()
        n_right = right_mask.sum()
        n_total = n_left + n_right
        
        if n_left < self.min_samples_leaf or n_right < self.min_samples_leaf:
            return 0
        
        # 方差减少准则
        parent_cate = compute_cate(np.ones(len(y), dtype=bool))
        left_cate = compute_cate(left_mask)
        right_cate = compute_cate(right_mask)
        
        # 纯度增加 = 节点 CATE 方差减少
        purity_gain = (
            (n_left / n_total) * abs(left_cate - parent_cate) +
            (n_right / n_total) * abs(right_cate - parent_cate)
        )
        return purity_gain
    
    def fit(self, X, treatment, y):
        # 使用 sklearn 的 DecisionTreeClassifier 作为基框架
        # 自定义分裂准则需要重写整个树构建逻辑
        self.tree = DecisionTreeClassifier(max_depth=self.max_depth)
        # 转换问题:预测 uplift 导向的目标
        uplift_proxy = y * (2 * treatment - 1)
        self.tree.fit(X, uplift_proxy)
    
    def predict_uplift(self, X):
        return self.tree.predict(X)

CUPL(Classifying Univariate Patterns of Lift)

原理:使用集成树方法来减少方差,同时引入随机化来提高稳定性。


深度学习方法

CEVAE(Counterfactual Variational Autoencoder)

架构:结合 VAE 和潜在变量模型来估计个体治疗效应。5

核心思想

  • 使用变分自编码器学习混淆因素的潜在表示
  • 结合倾向得分和结果模型进行双重估计
# CEVAE 的简化概念实现
class CEVAE(nn.Module):
    def __init__(self, input_dim, latent_dim=20):
        super().__init__()
        
        # 编码器:q(z|x, t, y)
        self.encoder_t0 = nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.ReLU(),
            nn.Linear(128, latent_dim * 2)  # mu, log_var
        )
        self.encoder_t1 = nn.Sequential(
            nn.Linear(input_dim, 128),
            nn.ReLU(),
            nn.Linear(128, latent_dim * 2)
        )
        
        # 倾向得分网络
        self.propensity_net = nn.Sequential(
            nn.Linear(input_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1),
            nn.Sigmoid()
        )
        
        # 结果网络
        self.outcome_net_t0 = nn.Sequential(
            nn.Linear(latent_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
        self.outcome_net_t1 = nn.Sequential(
            nn.Linear(latent_dim, 64),
            nn.ReLU(),
            nn.Linear(64, 1)
        )
    
    def forward(self, x, t):
        # 估计倾向得分
        prop = self.propensity_net(x)
        
        # 根据 treatment 选择编码器
        if t == 1:
            h = self.encoder_t1(x)
        else:
            h = self.encoder_t0(x)
        
        # 采样潜在变量
        mu, log_var = h[:, :20], h[:, 20:]
        z = mu + torch.randn_like(mu) * torch.exp(0.5 * log_var)
        
        # 预测结果
        y_pred = self.outcome_net_t1(z) if t == 1 else self.outcome_net_t0(z)
        
        return y_pred, prop, z

XGBoost uplift

使用现有的梯度提升框架,但结合专门的 uplift 目标函数:

# 带 uplift 目标函数的 XGBoost
import xgboost as xgb
 
class XGBoostUplift:
    def __init__(self):
        self.model_t1 = None
        self.model_t0 = None
        self.propensity_model = None
    
    def fit(self, X, treatment, y):
        # 倾向得分估计
        self.propensity_model = xgb.XGBClassifier()
        self.propensity_model.fit(X, treatment)
        propensity = self.propensity_model.predict_proba(X)[:, 1]
        
        # IPTW 加权的 T-Learner
        # 干预组的权重
        weight_t1 = treatment / (propensity + 1e-5)
        # 对照组的权重
        weight_t0 = (1 - treatment) / (1 - propensity + 1e-5)
        
        self.model_t1 = xgb.XGBRegressor()
        self.model_t1.fit(X[treatment==1], y[treatment==1], 
                          sample_weight=weight_t1[treatment==1])
        
        self.model_t0 = xgb.XGBRegressor()
        self.model_t0.fit(X[treatment==0], y[treatment==0],
                          sample_weight=weight_t0[treatment==0])
    
    def predict_uplift(self, X):
        return self.model_t1.predict(X) - self.model_t0.predict(X)

方法对比

方法优点缺点适用场景
S-Learner简单 不够敏感基线、初步探索
T-Learner分工明确需要大量数据两组样本均衡
X-Learner适应样本不平衡实现复杂一组样本明显少
DR-Learner双重稳健依赖倾向得分估计观测数据
Class Transformation简单、端到端假设样本平衡随机实验
Causal Tree可解释、统计性质好计算复杂度高需要可解释性
深度学习方法表达力强需要大量数据复杂非线性关系

实践建议

数据要求

  1. 随机实验数据最佳:保证
  2. 样本量:uplift 估计方差较大,通常需要较大的样本量
  3. 特征设计:包含可能影响响应和治疗的特征

模型选择

  1. 数据量小(< 10K):T-Learner 或 Class Transformation
  2. 数据量中等(10K - 100K):X-Learner 或 DR-Learner
  3. 数据量大(> 100K)+ 复杂关系:深度学习方法

注意事项

  1. 避免过拟合:使用交叉验证评估
  2. 检查倾向得分:确保没有极端的倾向值
  3. 业务验证:uplift 模型最终需要业务指标验证

参考资料

Footnotes

  1. Athey, S., & Imbens, G. (2016). Recursive partitioning for heterogeneous causal effects. PNAS, 113(27), 7353-7360. 2

  2. Künzel, S. R., Sekhon, J. S., Bickel, P. J., & Yu, B. (2019). Metalearners for estimating heterogeneous treatment effects using machine learning. Proceedings of the National Academy of Sciences, 116(10), 4156-4164.

  3. Kennedy, E. H. (2020). Towards optimal doubly robust estimation of heterogeneous causal effects. arXiv preprint arXiv:2004.14497.

  4. Jaskowski, M., & Jaroszewicz, S. (2012). Uplift modeling for clinical trial data. ICML Workshop on Clinical Data Analysis.

  5. Louizos, C., Shalit, U., Mooij, J. M., Sontag, D., Zemel, R., & Welling, M. (2017). Causal effect inference with deep latent-variable models. NeurIPS, 6446-6456.