第 9 章 AB 实验框架

对应源码: src/experiment/ab_test.py, src/experiment/metrics.py

模型做出的决策到底有没有效？只有对比实验才能给出可信答案。本章实现一个完整的 AB 实验框架：确定性分桶、统计检验、增量指标分析。

9.1 为什么必须做 AB 实验

没有实验就没有因果

"发券后用户转化率提升了 5%"——这能说明发券有效吗？

不能。也许这些用户本来就要买。也许是大盘流量变好了。也许是促销季的自然提升。

只有随机对照实验才能回答因果问题：在其他条件完全相同的情况下，发券的增量效果是多少？

AB 实验的核心逻辑：

1. 随机分桶：将用户随机分为实验组和对照组，保证两组在统计上"一样"
2. 差异化处理：实验组发券，对照组不发（或发安慰券）
3. 统计检验：比较两组的差异是否显著，排除偶然因素

9.2 Hash 分桶：确定性 + 均匀性

9.2.1 为什么不用随机数

用 random.choice(["control", "treatment"]) 分桶有两个致命问题：

1. 不确定性：同一用户每次请求可能分到不同组，污染实验结果
2. 不可复现：无法事后验证某用户属于哪个组

解决方案：用 user_id 的 hash 值做确定性分桶。

9.2.2 实现

# src/experiment/ab_test.py

class ABTestFramework:
    """基于 user_id hash 的确定性分桶."""

    _GROUP_RATIO_ATTRS = (
        ("control", "control_ratio"),   # 对照组 10%
        ("model", "model_ratio"),       # 模型组 45%
        ("rule", "rule_ratio"),         # 规则组 30%
        ("explore", "explore_ratio"),   # 探索组 15%
    )

    def __init__(
        self,
        config: CouponConfig | None = None,
        salt: str = "coupon_exp_v1",
    ) -> None:
        self.config = config or default_config
        self.salt = salt

        # 构建分桶边界 (累积比例)
        self._boundaries: list[tuple[str, float]] = []
        cumsum = 0.0
        for name, attr in self._GROUP_RATIO_ATTRS:
            cumsum += getattr(self.config, attr)
            self._boundaries.append((name, cumsum))

    def assign_group(self, user_id: str) -> str:
        """将单个用户确定性分配到实验组."""
        h = hashlib.md5(f"{self.salt}:{user_id}".encode()).hexdigest()
        bucket = int(h[:8], 16) / 0xFFFFFFFF  # → [0, 1]
        for name, upper in self._boundaries:
            if bucket < upper:
                return name
        return self._boundaries[-1][0]  # 浮点舍入兜底

关键设计点：

设计	原因
MD5 hash	均匀分布、计算快、确定性
salt 前缀	不同实验用不同 salt，避免分桶相关
取前 8 位十六进制	32 位精度（42 亿个桶），足够均匀
累积比例边界	支持任意比例的多组分桶

9.2.3 四组实验设计

我们的实验不是简单的二组对照，而是四组：

控制组 (10%)  — 不发券，作为因果推断的基线
模型组 (45%)  — 由 Uplift 模型 + 预算优化器决定是否发券和面额
规则组 (30%)  — 按生命周期规则发固定面额，作为模型的对比基线
探索组 (15%)  — 随机发券，收集训练数据，避免模型"近视"

为什么保留探索组？

模型只能从历史数据中学习。如果我们只按模型发券，那些被模型判为"不值得发"的用户永远不会收到券，模型也就永远学不到他们的真实响应。

探索组的随机发券打破了这种选择偏差，为模型提供无偏的训练样本。这是探索-利用权衡（Exploration-Exploitation）的经典应用。

9.2.4 验证分桶均匀性

# tests/test_optimizer.py

def test_group_ratio_approximately_correct(self, config):
    ab = ABTestFramework(config)
    user_ids = [f"user_{i}" for i in range(10000)]
    result = ab.assign_groups_batch(user_ids)

    counts = result["group"].value_counts(normalize=True)
    # 允许 3% 误差
    assert abs(counts.get("control", 0) - config.control_ratio) < 0.03
    assert abs(counts.get("model", 0) - config.model_ratio) < 0.03

9.3 效果分析与统计检验

9.3.1 核心指标

实验分析需要比较这些指标：

指标	含义	计算方式
转化率差	增量转化率	实验组 CVR - 对照组 CVR
增量人均 GMV	人均收入提升	实验组 avg_GMV - 对照组 avg_GMV
券 ROI	每元券带来多少增量 GMV	增量总 GMV / 总券成本
p-value	统计显著性	两比例 z 检验

9.3.2 两比例 z 检验

检验实验组和对照组的转化率差异是否显著：

$$z=\frac{{\hat{p}}_2-{\hat{p}}_1}{\sqrt{\hat{p}(1-\hat{p})(\frac{1}{n_1}+\frac{1}{n_2})}}$$

其中 $\hat{p}=\frac{x_1+x_2}{n_1+n_2}$ 是合并转化率。

# src/experiment/ab_test.py

@staticmethod
def _two_proportion_z_test(
    x1: int, n1: int, x2: int, n2: int
) -> float:
    """两比例 z 检验，返回双侧 p-value."""
    if n1 == 0 or n2 == 0:
        return 1.0
    p1 = x1 / n1
    p2 = x2 / n2
    p_pool = (x1 + x2) / (n1 + n2)
    se = np.sqrt(p_pool * (1 - p_pool) * (1 / n1 + 1 / n2))
    if se == 0:
        return 1.0
    z = (p2 - p1) / se
    return float(2 * (1 - stats.norm.cdf(abs(z))))

9.3.3 完整分析器

# src/experiment/ab_test.py

class ExperimentAnalyzer:
    """实验组 vs 对照组的增量效果与统计检验."""

    def analyze(self, data: pd.DataFrame) -> dict:
        result: dict = {"groups": {}, "comparisons": {}}

        # 分组汇总
        for group_name, gdf in data.groupby("group"):
            n = len(gdf)
            conversions = gdf["converted"].sum()
            result["groups"][group_name] = {
                "n": n,
                "conversions": int(conversions),
                "conversion_rate": conversions / n if n else 0.0,
                "avg_gmv": gdf["gmv"].mean(),
                "total_gmv": gdf["gmv"].sum(),
                "coupon_cost": gdf["coupon_amount"].sum(),
            }

        # 各实验组 vs 对照组
        control = result["groups"].get("control")
        if control is None or control["n"] == 0:
            return result

        ctrl_df = data[data["group"] == "control"]

        for group_name in ("model", "rule", "explore"):
            grp = result["groups"].get(group_name)
            if grp is None or grp["n"] == 0:
                continue

            exp_df = data[data["group"] == group_name]

            # 增量指标
            incremental_gmv = grp["avg_gmv"] - control["avg_gmv"]
            cost = grp["coupon_cost"]
            incremental_total_gmv = incremental_gmv * grp["n"]
            roi = incremental_total_gmv / cost if cost > 0 else float("inf")

            # 统计检验
            p_value = self._two_proportion_z_test(
                ctrl_df["converted"].sum(), control["n"],
                exp_df["converted"].sum(), grp["n"],
            )

            result["comparisons"][group_name] = {
                "incremental_cvr": grp["conversion_rate"] - control["conversion_rate"],
                "incremental_avg_gmv": incremental_gmv,
                "incremental_total_gmv": incremental_total_gmv,
                "coupon_roi": roi,
                "p_value": p_value,
                "significant_at_005": p_value < 0.05,
            }

        return result

9.4 Uplift 模型评估指标

AB 实验衡量的是整体策略效果，但我们还需要评估 Uplift 模型本身的排序能力。第 6 章训练了模型，这里给出配套的评估指标。

9.4.1 AUUC（Area Under the Uplift Curve）

类似 AUC 之于分类模型，AUUC 衡量 Uplift 模型的排序能力：

1. 按 uplift_score 降序排列用户
2. 逐步累积 Top-k% 用户的增量转化率
3. 曲线下面积越大，模型越能把"真正受益于发券"的用户排在前面

# src/experiment/metrics.py

def uplift_curve(y_true, uplift_scores, treatment):
    """按 uplift_score 降序排列，逐步累积增量转化率."""
    order = np.argsort(-uplift_scores)
    y_sorted = y_true[order]
    t_sorted = treatment[order]

    n = len(y_true)
    percentiles = np.arange(1, n + 1) / n

    cum_treat_conv = np.cumsum(y_sorted * t_sorted)
    cum_treat_n = np.cumsum(t_sorted)
    cum_ctrl_conv = np.cumsum(y_sorted * (1 - t_sorted))
    cum_ctrl_n = np.cumsum(1 - t_sorted)

    treat_cvr = np.divide(cum_treat_conv, cum_treat_n,
                          out=np.zeros_like(cum_treat_conv),
                          where=cum_treat_n > 0)
    ctrl_cvr = np.divide(cum_ctrl_conv, cum_ctrl_n,
                         out=np.zeros_like(cum_ctrl_conv),
                         where=cum_ctrl_n > 0)

    return percentiles, treat_cvr - ctrl_cvr


def auuc(y_true, uplift_scores, treatment) -> float:
    """Area Under the Uplift Curve."""
    pct, values = uplift_curve(y_true, uplift_scores, treatment)
    return float(np.trapezoid(values, pct))

9.4.2 Qini 系数

AUUC 的绝对值受数据集影响大，Qini 系数做了归一化：

$$\text{Qini}=\frac{{\text{AUUC}}_{\text{model}}-{\text{AUUC}}_{\text{random}}}{{\text{AUUC}}_{\text{ideal}}-{\text{AUUC}}_{\text{random}}}$$

• ${\text{AUUC}}_{\text{random}}=\text{overall\_uplift}\times 0.5$
• ${\text{AUUC}}_{\text{ideal}}$：用真实标签排序的上界

Qini ∈ [0, 1]，越接近 1 说明模型排序越接近理想排序。

9.4.3 Uplift@K

实际发券时我们不会给所有人发，只发给 Top-K%。Uplift@K 衡量的就是我们真正触达的那批人的增量效果：

def uplift_at_k(y_true, uplift_scores, treatment, k=0.1) -> float:
    """Top k% 用户的增量转化率."""
    pct, values = uplift_curve(y_true, uplift_scores, treatment)
    idx = max(0, np.searchsorted(pct, k) - 1)
    return float(values[idx])

指标选择指南

场景	推荐指标
模型迭代、离线评估	AUUC + Qini（整体排序能力）
预算受限、只发 Top-K	Uplift@K（实际触达用户的效果）
全量上线后评估	AB 实验的增量 CVR + 券 ROI

9.5 本章小结

• Hash 分桶保证同一用户每次分到同一组，确定性 + 均匀性
• 四组设计（对照 / 模型 / 规则 / 探索）兼顾因果推断和数据收集
• z 检验验证转化率差异的统计显著性，p < 0.05 拒绝原假设
• AUUC / Qini / Uplift@K 从不同角度评估 Uplift 模型排序能力
• 永远保留对照组——没有实验就没有因果

💡 思考题

1. 如果线上实验跑了 3 天后，发现模型组的 p-value = 0.08（不显著），你会做什么决策？是继续跑还是下线？需要考虑哪些因素？
2. 探索组的 15% 流量是固定的。如果模型已经很成熟，探索组的比例是否应该降低？这涉及到什么经典问题？
3. 我们的 z 检验假设样本量足够大（中心极限定理）。如果转化率很低（如 0.1%），小样本下应该用什么替代检验方法？