Promotion Effectiveness Evaluation with Causal ML

1. 解决的问题

促销活动期间销售额暴涨30%，这是促销的真实效果，还是"本来就会买的用户"恰好遇到了促销？

2. 核心算法逻辑

核心问题：促销活动期间销售额暴涨30%，这是促销的真实效果，还是"本来就会买的用户"恰好遇到了促销？如果不做促销，销售额会是多少？促销是否侵蚀了利润？

3. 业务应用场景

业务问题：Momcozy对新注册用户发20%首单优惠券，使用率为35%，使用后平均订单金额$80。团队认为"首单折扣很成功"。真实增量是多少？

naive分析（错误）： - 使用优惠券的用户平均消费$80 - 假设不用券会买$0 → 增量=$80 - 1000人使用 → 增量收入=$80,000

DML因果分析（正确）： 1. 找到"类似但未收到券"的用户作为对照组（Propensity Score Matching） 2. 控制变量：用户来源渠道、注册时间、浏览行为、 demographics 3. DML估计结果：

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• ROI：识别无效促销支出后，可削减20-40%的促销浪费，年节省数十万
• 难度：⭐⭐⭐⭐☆（4/5）— 因果推断概念门槛高，DML实现复杂
• 优先级：⭐⭐⭐⭐⭐（5/5）— 避免"促销幻觉"，确保每一分折扣都产生真实增量

7. 代码模板

代码块数量：1 · 路径：未检测到

"""
Promotion Effectiveness Evaluation — 促销效果评估
支持：DML估计CATE、Uplift Modeling、利润分析
"""

import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingRegressor, GradientBoostingClassifier
from sklearn.model_selection import cross_val_predict, KFold


class PromotionEffectDML:
    """基于DML的促销效果评估"""

    def __init__(self, n_splits=5):
        self.n_splits = n_splits
        self.model_y = GradientBoostingRegressor(n_estimators=100, random_state=42)
        self.model_t = GradientBoostingClassifier(n_estimators=100, random_state=42)
        self.theta = None  # CATE估计
        self.residual_y_model = GradientBoostingRegressor(n_estimators=100, random_state=42)
        self.residual_t_model = GradientBoostingRegressor(n_estimators=100, random_state=42)

    def fit(self, X, T, Y):
        """
        拟合DML模型估计CATE

        Args:
            X: 特征矩阵 (n, d)
            T: 处理变量 (n,) — 1=有促销, 0=无促销
            Y: 结果变量 (n,) — 购买金额/利润
        """
        n = len(Y)
        kf = KFold(n_splits=self.n_splits, shuffle=True, random_state=42)

        # 第一阶段：交叉拟合预测
        Y_hat = np.zeros(n)
        T_hat = np.zeros(n)

        for train_idx, test_idx in kf.split(X):
            X_train, X_test = X[train_idx], X[test_idx]
            Y_train, Y_test = Y[train_idx], Y[test_idx]
            T_train, T_test = T[train_idx], T[test_idx]

            # 拟合E[Y|X]
            self.model_y.fit(X_train, Y_train)
            Y_hat[test_idx] = self.model_y.predict(X_test)

            # 拟合E[T|X]
            self.model_t.fit(X_train, T_train)
            T_hat[test_idx] = self.model_t.predict_proba(X_test)[:, 1]

        # 计算残差
        Y_tilde = Y - Y_hat
        T_tilde = T - T_hat

        # 第二阶段：用残差估计CATE
        # theta(X) = E[Y_tilde / T_tilde | X] ≈ 用X预测Y_tilde/T_tilde
        # 简化：对每个样本直接计算局部效应，再用模型平滑
        local_effects = np.divide(Y_tilde, T_tilde,
                                  out=np.zeros_like(Y_tilde),

8. 论文来源

未自动抽取；请查看原始 Skill 卡片。