① 问题 ② 算法 ③ 场景 ④ 代码 ⑤ 关联 ⑥ 价值

STATE — 重尾指标鲁棒 A/B 方差减少：Student-t 回归调整（-70% 方差）

Skill-STATE-Robust-Variance-Reduction · 02-A_B实验

causalexperimentforecastingoptimization广告与投放WF-B 广告优化WF-F 动态定价WF-G Listing内容优化

收录于用户增长决策手册新品冷启动手册

年化 ROI30-60 万元

实现难度⭐⭐⭐☆☆

业务优先级⭐⭐⭐⭐⭐

业务视角

适用角色运营负责人 / 产品经理 · 广告优化师 · 选品负责人

适用平台Amazon Listing · TikTok 广告素材 · DTC 落地页

什么情况下用改了主图/标题/价格，不确定销量变化是改动导致的还是流量波动；两个方案团队各持己见，需要数据裁决

成功是什么样的每次改动都有 ≥95% 置信度的数据结论，好的改动快速全量，坏的及时止损

业务痛点

改了主图感觉好多了但不确定小范围测试结果好全量后没效果测试周期短结论不可靠

1. 解决的问题

电商 GMV / 订单量等指标天然重尾——极少数大客户的超大订单把方差撑得很高。

2. 核心算法逻辑

核心问题：电商 GMV / 订单量等指标天然重尾——极少数大客户的超大订单把方差撑得很高。CUPED 假设残差服从高斯分布，重尾数据下高斯假设失效，方差缩减效率大打折扣；CUPAC/MLRATE 用机器学习拟合协变量改进了均值预测，但同样没处理残差的非高斯性。

3. 业务应用场景

场景一：母婴新品 Listing AB 测试（GMV 重尾问题）

- 业务问题：新品奶粉上架后测试两种 Listing 方案（主图 A vs 主图 B），以 GMV 作为主指标。少数大客户（批量采购/月子中心）产生 10-50 倍于普通用户的订单，使 GMV 方差极高。传统 CUPED 需跑 3 周才显著； - 数据要求：实验期 GMV（$Y$）+ 实验前 14 天 GMV/浏览/加购等协变量（$X$），用户级别数据 - STATE 做法：以 XGBoost 拟合 $\hat{Y} = f(X)$，残差 $\varepsilon$ 用 EM 估 t 分布参数（典型 $\hat{\nu} \approx 3.5$），得到方差缩减后的调整指标 - 预期产出：方差

场景二：WF-B 广告竞价策略测试（点击/订单量重尾）

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

ROI：实验时长减半（3 周→1.5 周），母婴电商每实验节省 1.5 周迭代时间；年化可多跑 ~17 轮实验，隐性价值 30-60 万元（每轮实验加速决策带来的优化收益）
实施难度：⭐⭐⭐☆☆（需理解 EM 算法，但有成熟代码模板）
优先级：⭐⭐⭐⭐⭐（重尾指标是电商 A/B 实验的普遍痛点，直接缩短实验周期）
评估依据：美团真实数据验证 70.5%/80.7% 方差减少；母婴 GMV 指标天然重尾（大客户聚合效应），效果复现概率高

7. 代码模板

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

"""
STATE Variance Reduction — Student's t Regression Adjustment
arXiv:2407.16337 | 美团 2024
纯标准库 + numpy/scipy 实现
"""
from __future__ import annotations

import math
import numpy as np
from dataclasses import dataclass, field
from typing import Optional, Tuple
from scipy import stats


# ─── 数据结构 ─────────────────────────────────────────────────────────────────

@dataclass
class HeavyTailMetrics:
    """实验指标数据容器"""
    metric_values: np.ndarray        # 实验期指标 Y（GMV / 订单量）
    pre_exp_covariates: np.ndarray   # 实验前协变量 X（可多列）
    treatment: np.ndarray            # 处理标记：1=实验组，0=对照组

    def __post_init__(self):
        n = len(self.metric_values)
        assert len(self.treatment) == n, "treatment length mismatch"
        if self.pre_exp_covariates.ndim == 1:
            self.pre_exp_covariates = self.pre_exp_covariates.reshape(-1, 1)
        assert len(self.pre_exp_covariates) == n, "covariate length mismatch"

    @property
    def treatment_idx(self) -> np.ndarray:
        return self.treatment == 1

    @property
    def control_idx(self) -> np.ndarray:
        return self.treatment == 0


@dataclass
class ATEResult:
    """ATE 估计结果"""
    ate: float                       # 平均处理效应
    variance: float                  # 估计量方差
    std_error: float                 # 标准误
    p_value: float                   # 双侧 p-value
    ci_lower: float                  # 95% 置信区间下界
    ci_upper: float                  # 95% 置信区间上界
    t_df: Optional[float] = None     # t 分布自由度（STATE 专属）
    t_scale: Optional[float] = None  # t 分布尺度（STATE 专属）
    variance_reduction_vs_raw: float = 0.0  # vs 原始方差的缩减率

    def __str__(self) -> str:
        sig = "✓ 显著" if self.p_value < 0.05 else "✗ 不显著"
        return (
            f"ATE={self.ate:.4f}  SE={self.std_error:.4f}  "
            f"p={self.p_value:.4f}  {sig}\n"
            f"95% CI=[{self.ci_lower:.4f}, {self.ci_upper:.4f}]\n"
            f"方差缩减={self.variance_reduction_vs_raw:.1%}"
            + (f"  t_df={self.t_df:.2f}  t_scale={self.t_scale:.4f}" if self.t_df else "")

8. 论文来源

2407.16337