Switchback 实验设计 - 数据驱动的双边市场实验

1. 解决的问题

同一海外仓为 Shopify/Amazon/TikTok Shop 多渠道发货,测试"AI 波次合并算法"是否降低拣货时长

2. 核心算法逻辑

在传统 A/B 难以适用的双边市场(物流仓配、动态定价、平台撮合)场景下,Switchback 实验通过对单一聚合单元随时间反复切换处理/控制状态来估计因果效应。本论文给出 MSE 偏差方差四因子分解 与 Empirical Bayes 设计选择框架,自动选最优切换方案。

3. 业务应用场景

- 业务问题:同一海外仓为 Shopify/Amazon/TikTok Shop 多渠道发货,测试"AI 波次合并算法"是否降低拣货时长。仓库内强 SUTVA 违反——一批订单占用传送带影响下一批。 - 数据要求:逐订单拣货耗时日志 + 班次时间戳 + 算法启用标记 - Switchback 配置: - 切换粒度:4 小时(单班次) - 处理:开启 AI 合并 vs 现有规则 - Carryover τ:1-2 班次(传送带预热) - 原则应对:按早/中/夜班平衡周期性 + 区间≥2班次 + ±15min 随机边界 - 业务价值:相比传统集群随机实验(不可行,仓库唯一),Switchback

- 业务问题:测试"需求感知动态运费"策略(旺季涨价/淡季折扣)对 7 日复购率 × 客单价 = LTV 增量的净影响。买家抢购影响库存可见性,SUTVA 违反。 - 数据要求:用户行为日志 + 订单日志 + 运费策略状态 - Switchback 配置: - 切换粒度:1 天 - Carryover τ:7-14 天(购买习惯形成) - 区间长度:14 天(≥τ_max) - Empirical Bayes:用历史节促 CEC 数据构建先验,自动选最优设计 - 业务价值:动态定价策略验证准确性提升 33%,以中型站月 GMV 1000 万元计,价格优化 GMV 增量 2-5%/年 = 240

4. 输入数据要求

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

5. 输出结果

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

6. 业务价值 / ROI

• 难处:Empirical Bayes 设计需要历史 CEC 数据(没有就先做粗设计积累)
• 难处:HT 估计的方差计算需要 Newey-West 校正,工程实现稍复杂
• 易处:第三方 R 复现代码可参考

7. 代码模板

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

"""
Switchback Experiment 最小骨架
论文 arXiv:2406.06768 (Xiong et al., 2024)
第三方 R 复现: https://github.com/QianglinSIMON/SwitchMDP
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Dict, List, Tuple

import numpy as np


@dataclass
class SwitchbackConfig:
    n_periods: int = 48
    avg_interval_len: int = 4
    balance_periodicity: bool = True
    randomize_boundaries: bool = True


def generate_switchback_assignment(cfg: SwitchbackConfig, seed: int = 42) -> np.ndarray:
    rng = np.random.default_rng(seed)
    intervals: List[Tuple[int, int, int]] = []
    t = 0
    while t < cfg.n_periods:
        length = max(1, rng.poisson(cfg.avg_interval_len))
        if cfg.randomize_boundaries:
            length += rng.integers(-1, 2)
        treatment = (len(intervals) % 2) if cfg.balance_periodicity else int(rng.integers(2))
        intervals.append((t, min(t + length, cfg.n_periods), treatment))
        t += length

    W = np.zeros(cfg.n_periods, dtype=int)
    for start, end, w in intervals:
        W[start:end] = w
    return W


def ht_estimator(outcomes: np.ndarray, W: np.ndarray, p: float = 0.5) -> Dict[str, float]:
    treated = outcomes[W == 1] / p
    control = outcomes[W == 0] / (1 - p)
    gate_hat = float(treated.mean() - control.mean())
    se = float(np.sqrt(np.var(treated) / max(len(treated), 1) + np.var(control) / max(len(control), 1)))
    return {"GATE": gate_hat, "SE": se, "CI_low": gate_hat - 1.96 * se, "CI_high": gate_hat + 1.96 * se}


def empirical_bayes_design(historical_cecs: np.ndarray, candidate_configs: List[SwitchbackConfig]) -> SwitchbackConfig:
    best_cfg = candidate_configs[0]
    best_mse = float("inf")
    rng = np.random.default_rng(0)
    for cfg in candidate_configs:
        mse_samples = []
        for cec in historical_cecs:
            W = generate_switchback_assignment(cfg, seed=int(rng.integers(0, 100000)))
            Y = rng.standard_normal(cfg.n_periods) + cec * W
            est = ht_estimator(Y, W)
            mse_samples.append((est["GATE"] - cec) ** 2)
        mse = float(np.mean(mse_samples))
        if mse < best_mse:
            best_mse = mse

8. 论文来源

• 2406.06768