experiment_platform.py

1"""对应正文第 3 章 §3.3.2「A/B实验平台」中列举的五大子系统。23正文原话：「一个典型实验平台包含：实验配置管理，流量分桶系统，指标计算系统，4实验分析系统，风险控制系统。为了提高流量利用率，许多公司还会采用分层实验5（Layered Experimentation）架构……不同层之间互不干扰，从而实现同一批用户流量6被多个实验同时复用。」78本实现做了什么9---------------10把这五个子系统各写成一个进程内 `[MOCK]` 组件，并串成一次完整的实验生命周期：111. **实验配置管理（ConfigRegistry）**：实验以 `ExperimentConfig` 描述（所属层、12   桶区间、参数、核心/诊断指标与护栏阈值）。注册时做三类静态校验——13   (a) 层内**互斥**：同层实验的桶区间不得重叠；14   (b) 层内流量不得超过 100 个桶；15   (c) 参数必须落在该层声明的参数空间内（防止一个层里两个实验改同一个参数）。162. **实验配置下发（ConfigDistributor）**：带版本号的配置灰度推送到多台 `[MOCK]`17   在线服务实例，模拟推送延迟与单实例推送失败，并监控「配置版本一致率」——18   版本不一致期间该实例仍按旧配置分流，是线上实验「脏流量」的常见来源。193. **流量分桶系统（BucketService）**：md5(层 salt + uid) mod 100，层内互斥、20   层间正交，同一用户流量被多层实验同时复用。214. **指标计算系统（MetricPipeline）**：按 (实验, 组) 聚合 `[MOCK]` 曝光/行为日志，22   算核心指标（CTR、人均时长）与诊断指标（hate 率、P99 延迟）。235. **实验分析系统（AnalysisService）**：Welch t 检验 / 两比例 z 检验 + 95% 置信区间，24   给出显著性结论。256. **风险控制系统（RiskControl）**：实时巡检护栏指标，触发阈值即自动「止损」26   （把实验流量拉回 0 并回滚配置版本）。2728关键公式29---------30    bucket(uid, layer) = md5(f"{salt}:{uid}") mod 10031    Welch t = (x1-x2)/sqrt(s1^2/n1+s2^2/n2)32    两比例 z = (p1-p2)/sqrt(p_pool(1-p_pool)(1/n1+1/n2))33    配置版本一致率 = #{实例 version == 最新 version} / #实例34"""3536from __future__ import annotations3738import hashlib39import math40import random41from collections import defaultdict42from dataclasses import dataclass, field434445# ==========================================================================46# 统计工具（纯标准库）47# ==========================================================================48def norm_cdf(x: float) -> float:49    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))505152def welch_ttest(a: list[float], b: list[float]) -> dict:53    """双样本 Welch t 检验（方差不等），大样本下用正态近似算 p 与 CI。"""54    n1, n2 = len(a), len(b)55    m1, m2 = sum(a) / n1, sum(b) / n256    v1 = sum((x - m1) ** 2 for x in a) / (n1 - 1)57    v2 = sum((x - m2) ** 2 for x in b) / (n2 - 1)58    se = math.sqrt(v1 / n1 + v2 / n2)59    t = (m1 - m2) / se60    p = 2 * (1 - norm_cdf(abs(t)))61    return {"mean_exp": m1, "mean_base": m2, "diff": m1 - m2, "t": t, "p": p,62            "ci_low": (m1 - m2) - 1.959964 * se, "ci_high": (m1 - m2) + 1.959964 * se,63            "lift_pct": 100.0 * (m1 / m2 - 1)}646566def two_proportion_ztest(c1: int, n1: int, c2: int, n2: int) -> dict:67    """两比例 z 检验（CTR 这类比例指标）。"""68    p1, p2 = c1 / n1, c2 / n269    pp = (c1 + c2) / (n1 + n2)70    se_pool = math.sqrt(pp * (1 - pp) * (1 / n1 + 1 / n2))71    se_un = math.sqrt(p1 * (1 - p1) / n1 + p2 * (1 - p2) / n2)72    z = (p1 - p2) / se_pool if se_pool > 0 else 0.073    return {"p1": p1, "p2": p2, "z": z, "p": 2 * (1 - norm_cdf(abs(z))),74            "ci_low": (p1 - p2) - 1.959964 * se_un,75            "ci_high": (p1 - p2) + 1.959964 * se_un,76            "lift_pct": 100.0 * (p1 / p2 - 1) if p2 > 0 else float("nan")}777879# ==========================================================================80# 子系统 1：实验配置管理81# ==========================================================================82@dataclass83class ExperimentConfig:84    """一个实验的完整配置。buckets 为该实验在所属层占据的桶号集合。"""8586    exp_id: str87    layer: str88    buckets: range                      # 层内互斥地占据这些桶89    params: dict                        # 下发给在线服务的策略参数90    core_metrics: tuple[str, ...]91    guardrails: dict                    # 指标名 -> 允许的最大相对恶化（负向，%）92    owner: str = "rec-algo"93    status: str = "pending"             # pending / running / stopped / rolled_back949596@dataclass97class LayerSpec:98    """层定义：salt 决定层间正交；param_space 限定本层能改哪些参数。"""99100    name: str101    salt: str102    param_space: tuple[str, ...]103    n_buckets: int = 100104105106class ConfigRegistry:107    """[MOCK] 实验配置管理：注册 + 三类静态校验（互斥 / 容量 / 参数空间）。"""108109    def __init__(self, layers: list[LayerSpec]) -> None:110        self.layers = {ly.name: ly for ly in layers}111        self.experiments: dict[str, ExperimentConfig] = {}112        self.version = 0113114    def _occupied(self, layer: str) -> set[int]:115        occ: set[int] = set()116        for e in self.experiments.values():117            if e.layer == layer and e.status == "running":118                occ |= set(e.buckets)119        return occ120121    def register(self, cfg: ExperimentConfig) -> tuple[bool, str]:122        if cfg.layer not in self.layers:123            return False, f"层 {cfg.layer} 不存在"124        ly = self.layers[cfg.layer]125        bad = [k for k in cfg.params if k not in ly.param_space]126        if bad:127            return False, f"参数 {bad} 不在层 {cfg.layer} 的参数空间 {ly.param_space} 内"128        occ = self._occupied(cfg.layer)129        clash = sorted(occ & set(cfg.buckets))130        if clash:131            return False, (f"层内互斥冲突：桶 {clash[0]}~{clash[-1]} 已被同层实验占用"132                           f"（共 {len(clash)} 个桶）")133        if len(occ) + len(cfg.buckets) > ly.n_buckets:134            return False, f"层 {cfg.layer} 流量超额：已用 {len(occ)}% + 申请 {len(cfg.buckets)}%"135        cfg.status = "running"136        self.experiments[cfg.exp_id] = cfg137        self.version += 1138        return True, f"注册成功，占 {len(cfg.buckets)}% 流量，配置版本 -> v{self.version}"139140    def stop(self, exp_id: str, reason: str) -> None:141        self.experiments[exp_id].status = "stopped"142        self.version += 1143        print(f"      [ConfigRegistry] 实验 {exp_id} 已置为 stopped（{reason}），"144              f"配置版本 -> v{self.version}")145146147# ==========================================================================148# 子系统 2：实验配置下发149# ==========================================================================150@dataclass151class ServiceInstance:152    """[MOCK] 一台在线推荐服务实例，持有自己的配置版本快照。"""153154    host: str155    version: int = 0156    pushed_at_s: float = 0.0157158159class ConfigDistributor:160    """[MOCK] 配置灰度下发：分批推送 + 推送延迟 + 单实例失败。"""161162    def __init__(self, n_instances: int = 20, fail_rate: float = 0.10) -> None:163        self.instances = [ServiceInstance(f"rec-online-{i:02d}") for i in range(n_instances)]164        self.fail_rate = fail_rate165166    def push(self, version: int, batch_ratio: float = 0.25) -> list[tuple[float, float]]:167        """按 batch_ratio 分批推送，返回 [(时刻秒, 版本一致率)] 曲线。"""168        order = list(range(len(self.instances)))169        random.shuffle(order)170        batch = max(1, int(len(order) * batch_ratio))171        curve: list[tuple[float, float]] = []172        t = 0.0173        for start in range(0, len(order), batch):174            t += random.uniform(1.5, 3.0)                 # 批次间隔175            for idx in order[start:start + batch]:176                if random.random() >= self.fail_rate:     # 可能推送失败177                    self.instances[idx].version = version178                    self.instances[idx].pushed_at_s = t179            curve.append((t, self.consistency(version)))180        return curve181182    def consistency(self, version: int) -> float:183        ok = sum(1 for s in self.instances if s.version == version)184        return ok / len(self.instances)185186    def stale_hosts(self, version: int) -> list[str]:187        return [s.host for s in self.instances if s.version != version]188189190# ==========================================================================191# 子系统 3：流量分桶192# ==========================================================================193class BucketService:194    """[MOCK] 流量分桶系统：层内互斥、层间正交（不同 salt）。"""195196    def __init__(self, registry: ConfigRegistry) -> None:197        self.reg = registry198199    @staticmethod200    def bucket(uid: int, salt: str, n: int = 100) -> int:201        h = hashlib.md5(f"{salt}:{uid}".encode()).hexdigest()202        return int(h, 16) % n203204    def assign(self, uid: int) -> dict[str, tuple[str, str]]:205        """返回 {层名: (exp_id, 'exp'|'base')}；未命中任何实验则落在该层 base。"""206        out: dict[str, tuple[str, str]] = {}207        for name, ly in self.reg.layers.items():208            b = self.bucket(uid, ly.salt, ly.n_buckets)209            hit = ("__base__", "base")210            for e in self.reg.experiments.values():211                if e.layer == name and e.status == "running" and b in e.buckets:212                    hit = (e.exp_id, "exp")213                    break214            out[name] = hit215        return out216217218# ==========================================================================219# 子系统 4：指标计算220# ==========================================================================221@dataclass222class GroupAgg:223    """一个 (实验, 组) 的指标累加器。"""224225    imps: int = 0226    clicks: int = 0227    hates: int = 0228    users: set = field(default_factory=set)229    watch_by_user: dict = field(default_factory=lambda: defaultdict(float))230    latencies: list = field(default_factory=list)231232    def ctr(self) -> float:233        return self.clicks / self.imps if self.imps else 0.0234235    def hate_rate(self) -> float:236        return self.hates / self.imps if self.imps else 0.0237238    def watch_samples(self) -> list[float]:239        return [self.watch_by_user[u] for u in sorted(self.users)]240241    def p99_latency(self) -> float:242        if not self.latencies:243            return 0.0244        s = sorted(self.latencies)245        return s[min(len(s) - 1, int(0.99 * len(s)))]246247248class MetricPipeline:249    """[MOCK] 指标计算系统：消费曝光/行为日志，按 (实验, 组) 聚合。"""250251    def __init__(self) -> None:252        self.agg: dict[tuple[str, str], GroupAgg] = defaultdict(GroupAgg)253254    def consume(self, exp_id: str, group: str, uid: int, imps: int, clicks: int,255                hates: int, watch_s: float, latency_ms: float) -> None:256        g = self.agg[(exp_id, group)]257        g.imps += imps258        g.clicks += clicks259        g.hates += hates260        g.users.add(uid)261        g.watch_by_user[uid] += watch_s262        g.latencies.append(latency_ms)263264265# ==========================================================================266# 子系统 5 / 6：实验分析 + 风险控制267# ==========================================================================268class AnalysisService:269    """[MOCK] 实验分析系统：显著性检验 + 置信区间 + 结论。"""270271    @staticmethod272    def analyze(exp: GroupAgg, base: GroupAgg) -> dict:273        ctr = two_proportion_ztest(exp.clicks, exp.imps, base.clicks, base.imps)274        wt = welch_ttest(exp.watch_samples(), base.watch_samples())275        return {"ctr": ctr, "watch_time": wt}276277278class RiskControl:279    """[MOCK] 风险控制系统：护栏指标超阈值自动止损（kill switch）。"""280281    def __init__(self, registry: ConfigRegistry, distributor: ConfigDistributor) -> None:282        self.reg = registry283        self.dist = distributor284285    def inspect(self, exp_id: str, exp: GroupAgg, base: GroupAgg) -> list[str]:286        cfg = self.reg.experiments[exp_id]287        alarms: list[str] = []288        observed = {289            "hate_rate": (exp.hate_rate(), base.hate_rate(), "越低越好"),290            "p99_latency_ms": (exp.p99_latency(), base.p99_latency(), "越低越好"),291            "ctr": (exp.ctr(), base.ctr(), "越高越好"),292        }293        for name, limit_pct in cfg.guardrails.items():294            o, b, direction = observed[name]295            rel = 100.0 * (o / b - 1) if b else 0.0296            worse = rel if direction == "越高越好" else -rel297            if worse < limit_pct:298                alarms.append(f"{name}: {o:.4f} vs base {b:.4f}（相对 {rel:+.2f}%），"299                              f"突破护栏 {limit_pct:+.1f}%")300        if alarms:301            self.reg.stop(exp_id, reason="护栏指标恶化")302            self.dist.push(self.reg.version)303        return alarms304305306# ==========================================================================307# [MOCK] 线上流量308# ==========================================================================309def simulate_user_day(uid: int, assignment: dict[str, tuple[str, str]],310                      registry: ConfigRegistry) -> dict:311    """[MOCK] 一个用户一天的汇总行为；每层命中的实验按自身效应乘性叠加。"""312    rng = random.Random(uid * 104729 + 7)313    imps = max(3, int(rng.gauss(26, 9)))314    ctr, wt_per_imp, hate, lat = 0.072, 1.55, 0.0045, 38.0315    for layer, (exp_id, grp) in assignment.items():316        if grp != "exp":317            continue318        eff = registry.experiments[exp_id].params319        ctr *= eff.get("ctr_mult", 1.0)320        wt_per_imp *= eff.get("wt_mult", 1.0)321        hate *= eff.get("hate_mult", 1.0)322        lat *= eff.get("latency_mult", 1.0)323    clicks = sum(1 for _ in range(imps) if rng.random() < ctr)324    hates = sum(1 for _ in range(imps) if rng.random() < hate)325    watch = sum(rng.expovariate(1.0 / max(wt_per_imp, 1e-6)) for _ in range(clicks + 1))326    return {"imps": imps, "clicks": clicks, "hates": hates, "watch_s": watch,327            "latency_ms": lat * rng.lognormvariate(0.0, 0.35)}328329330def main() -> None:331    random.seed(42)332    print("=" * 86)333    print("§3.3.2 A/B实验平台：配置管理 / 配置下发 / 流量分桶 / 指标计算 / 实验分析 / 风险控制")334    print("=" * 86)335336    layers = [337        LayerSpec("L1-召回层", "salt_recall_2024", ("recall_source", "ctr_mult", "wt_mult")),338        LayerSpec("L2-排序层", "salt_rank_2024",339                  ("model", "ctr_mult", "wt_mult", "latency_mult", "hate_mult")),340        LayerSpec("L3-重排层", "salt_rerank_2024", ("diversity_lambda", "ctr_mult", "hate_mult")),341    ]342    reg = ConfigRegistry(layers)343    dist = ConfigDistributor(n_instances=20, fail_rate=0.10)344345    # ---------------- (1) 配置管理：注册与校验 ----------------346    print("\n[1] 实验配置管理：注册 + 静态校验（层内互斥 / 容量 / 参数空间）")347    todo = [348        ExperimentConfig("exp_recall_i2i", "L1-召回层", range(90, 100),349                         {"recall_source": "i2i_v3", "ctr_mult": 1.03, "wt_mult": 1.01},350                         ("ctr",), {"hate_rate": -5.0}),351        ExperimentConfig("exp_rank_mmoe", "L2-排序层", range(85, 95),352                         {"model": "mmoe_v2", "ctr_mult": 1.05, "wt_mult": 1.04,353                          "latency_mult": 1.10},354                         ("ctr", "watch_time"), {"hate_rate": -5.0, "p99_latency_ms": -20.0}),355        # 与 exp_rank_mmoe 的桶区间重叠 -> 应被互斥校验拒绝356        ExperimentConfig("exp_rank_dcn", "L2-排序层", range(92, 100),357                         {"model": "dcn_v2", "ctr_mult": 1.01},358                         ("ctr",), {"hate_rate": -5.0}),359        # 改了不属于本层的参数 -> 应被参数空间校验拒绝360        ExperimentConfig("exp_rerank_bad", "L3-重排层", range(90, 100),361                         {"model": "rerank_mmr", "ctr_mult": 1.0},362                         ("ctr",), {"hate_rate": -5.0}),363        ExperimentConfig("exp_rerank_mmr", "L3-重排层", range(90, 100),364                         {"diversity_lambda": 0.3, "ctr_mult": 0.995, "hate_mult": 0.80},365                         ("ctr",), {"hate_rate": -5.0}),366        # 激进策略：CTR 涨但 hate 率与 P99 延迟双双恶化 -> 应被风控止损367        ExperimentConfig("exp_rank_aggressive", "L2-排序层", range(75, 85),368                         {"model": "ctr_only_v9", "ctr_mult": 1.12, "wt_mult": 0.93,369                          "hate_mult": 1.60, "latency_mult": 1.45},370                         ("ctr",), {"hate_rate": -5.0, "p99_latency_ms": -20.0}),371    ]372    print(f"  {'实验':>20} | {'层':>10} | {'桶区间':>9} | 结果")373    print("  " + "-" * 82)374    for cfg in todo:375        ok, msg = reg.register(cfg)376        rng_s = f"[{cfg.buckets.start},{cfg.buckets.stop - 1}]"377        print(f"  {cfg.exp_id:>20} | {cfg.layer:>10} | {rng_s:>9} | "378              f"{'✓ ' if ok else '✗ '}{msg}")379380    # ---------------- (2) 配置灰度下发 ----------------381    print(f"\n[2] 实验配置下发：当前配置版本 v{reg.version}，灰度推送到 20 台在线实例")382    curve = dist.push(reg.version, batch_ratio=0.25)383    print(f"  {'推送时刻(s)':>11} | {'版本一致率':>10} | 说明")384    print("  " + "-" * 56)385    for t, c in curve:386        note = "推送完成" if c == 1.0 else "部分实例仍按旧配置分流（脏流量）"387        print(f"  {t:>11.2f} | {100 * c:>9.1f}% | {note}")388    stale = dist.stale_hosts(reg.version)389    print(f"  最终一致率 {100 * dist.consistency(reg.version):.1f}%，"390          f"滞后实例 {len(stale)} 台：{stale if stale else '无'}")391    if stale:392        print("  >>> 配置下发未对齐期间，这些实例上的用户分组与指标归因都不可信，")393        print("      实验平台必须在「配置生效时间」之后才开始统计，否则实验组被旧配置污染。")394395    # ---------------- (3) 流量分桶与流量复用 ----------------396    print("\n[3] 流量分桶系统：层内互斥 + 层间正交 => 同一批用户被多个实验复用")397    bs = BucketService(reg)398    n_users = 60_000399    hit_count = defaultdict(int)       # 命中实验个数 -> 用户数400    per_exp = defaultdict(int)401    assigns: dict[int, dict] = {}402    for u in range(n_users):403        a = bs.assign(u)404        assigns[u] = a405        k = sum(1 for _, (_, g) in a.items() if g == "exp")406        hit_count[k] += 1407        for _, (eid, g) in a.items():408            if g == "exp":409                per_exp[eid] += 1410    print(f"  {'同时命中的实验数':>16} | {'用户数':>8} | {'占比':>8}")411    print("  " + "-" * 40)412    for k in sorted(hit_count):413        print(f"  {k:>16d} | {hit_count[k]:>8d} | {100.0 * hit_count[k] / n_users:>7.2f}%")414    exp_total = sum(per_exp.values())415    layer_use = defaultdict(int)416    for cfg in reg.experiments.values():417        layer_use[cfg.layer] += len(cfg.buckets)418    print(f"  实验覆盖人次合计 = {exp_total}，物理用户 = {n_users}，"419          f"人均参与实验 {exp_total / n_users:.2f} 个")420    print("  分层后各层流量占用：" + "，".join(421        f"{lid} {pct}%" for lid, pct in sorted(layer_use.items())))422    print(f"  若不分层，这 {len(reg.experiments)} 个实验要在单层里互斥瓜分 "423          f"{sum(layer_use.values())}% 流量；分层后每层都能独立用满 100%，"424          f"这正是正文说的「同一批用户流量被多个实验同时复用」")425    print(f"  {'实验':>20} | {'用户数':>8} | {'实际占比':>8} | {'设计占比':>8}")426    print("  " + "-" * 56)427    for eid, cfg in reg.experiments.items():428        print(f"  {eid:>20} | {per_exp[eid]:>8d} | "429              f"{100.0 * per_exp[eid] / n_users:>7.2f}% | {len(cfg.buckets):>7d}%")430431    # ---------------- (4) 指标计算 ----------------432    print("\n[4] 指标计算系统：消费一天的 [MOCK] 曝光/行为日志")433    mp = MetricPipeline()434    for u in range(n_users):435        a = assigns[u]436        day = simulate_user_day(u, a, reg)437        for layer, (eid, grp) in a.items():438            key_exp = eid if grp == "exp" else f"__base__::{layer}"439            mp.consume(key_exp, grp, u, day["imps"], day["clicks"], day["hates"],440                       day["watch_s"], day["latency_ms"])441    print(f"  {'统计单元':>26} | {'曝光':>9} | {'CTR':>7} | {'人均时长':>9} | "442          f"{'hate率':>7} | {'P99(ms)':>8}")443    print("  " + "-" * 86)444    for (key, grp), g in sorted(mp.agg.items()):445        wt = sum(g.watch_by_user.values()) / len(g.users)446        print(f"  {key[:26]:>26} | {g.imps:>9d} | {100 * g.ctr():>6.3f}% | "447              f"{wt:>9.3f} | {100 * g.hate_rate():>6.3f}% | {g.p99_latency():>8.2f}")448449    # ---------------- (5) 实验分析 ----------------450    print("\n[5] 实验分析系统：各实验 vs 同层 base（Welch t / 两比例 z，95% CI）")451    print(f"  {'实验':>20} | {'指标':>10} | {'Base':>9} | {'Exp':>9} | "452          f"{'相对提升':>9} | {'p-value':>9} | {'95% CI(绝对)':>24} | 结论")453    print("  " + "-" * 122)454    results = {}455    for eid, cfg in reg.experiments.items():456        e = mp.agg[(eid, "exp")]457        b = mp.agg[(f"__base__::{cfg.layer}", "base")]458        r = AnalysisService.analyze(e, b)459        results[eid] = (e, b, r)460        for mname, key, scale, unit in (("CTR", "ctr", 100.0, "%"),461                                        ("人均时长", "watch_time", 1.0, "s")):462            d = r[key]463            lo, hi = d["ci_low"] * scale, d["ci_high"] * scale464            base_v = (d["p2"] if key == "ctr" else d["mean_base"]) * scale465            exp_v = (d["p1"] if key == "ctr" else d["mean_exp"]) * scale466            sig = "显著正向" if d["p"] < 0.05 and d["lift_pct"] > 0 else (467                "显著负向" if d["p"] < 0.05 else "不显著")468            print(f"  {eid:>20} | {mname:>10} | {base_v:>8.4f}{unit} | {exp_v:>8.4f}{unit} | "469                  f"{d['lift_pct']:>+8.2f}% | {d['p']:>9.2e} | "470                  f"[{lo:>+8.4f}, {hi:>+8.4f}]{unit:>2} | {sig}")471472    # ---------------- (6) 风险控制 ----------------473    print("\n[6] 风险控制系统：巡检护栏指标，超阈值自动止损")474    for eid, (e, b, _) in results.items():475        alarms = RiskControl(reg, dist).inspect(eid, e, b)476        if alarms:477            print(f"  ✗ {eid}：触发 {len(alarms)} 条护栏告警")478            for a in alarms:479                print(f"      - {a}")480        else:481            print(f"  ✓ {eid}：护栏全部通过，允许继续放量")482    print("\n  实验最终状态：")483    for eid, cfg in reg.experiments.items():484        print(f"    {eid:>20} -> {cfg.status}")485    print(f"  配置版本已推进到 v{reg.version}，"486          f"止损后的一致率 {100 * dist.consistency(reg.version):.1f}%")487    print("\n  >>> 正文：实验平台由配置管理、流量分桶、指标计算、实验分析、风险控制五部分组成；")488    print("      exp_rank_aggressive 正是正文反复提醒的情形——核心指标 CTR 显著涨，")489    print("      但诊断指标（hate 率、P99 延迟）同时恶化，必须由风控系统拦住而非放量。")490491492if __name__ == "__main__":493    main()