mixrank_pipeline.py

1"""2混排模块端到端链路（第 8 章 §8.1 多业务混排问题，图「混排模块内部处理流程」）34对应正文概念5------------6正文把混排模块拆成四个核心环节：7    混排前处理（§8.2） → 混排序列生成与剪枝（§8.3） → 混排多业务排序（§8.4/§8.5） → 混排后处理（§8.6）8输入是推荐 Router 透传的多个业务排序列表（正文记作 {A1..A4}, ..., {D1..D4}），9输出是最终曝光给用户的一条融合序列。1011本实现做了什么12--------------13用一个可运行的 Python 链路把四个环节串起来，每一环都打印中间产物，便于读者看到14「一条请求是怎么从 4 个业务列表变成 1 条曝光序列的」：151. 前处理：保序回归（PAVA）校准各业务 pxtr + Min-Max 归一化，解决量纲不一致；162. 序列生成：在固定的短视频骨架上枚举广告/直播插入位置，并用约束规则剪枝；173. 序列排序：统一价值分 × 位置衰减因子 γ_i = e^{-λ(i-1)}，选 Top-1；184. 后处理：Load 调控（广告占比目标区间）+ 运营强插 + 保量/降权。1920关键公式21--------22    校准（IR/PAVA）:  r_k = Σ_{i∈B_k} y_i / |B_k|，相邻违序时合并23                     r_new = (|B_k| r_k + |B_{k+1}| r_{k+1}) / (|B_k| + |B_{k+1}|)24    归一化（Min-Max）: x' = (x - x_min) / (x_max - x_min)25    统一价值:          Value(item) = 兑换系数 × 业务原始价值26    序列价值:          Score_seq = Σ_{i=1..L} γ_i · Value(item_i),  γ_i = e^{-λ(i-1)}27    Load 调控:         score_final = score_raw + u_t2829[MOCK] 本文件中的 BusinessRecSys（各业务推荐系统）、Router（推荐路由）、30       ExposureLogStream（曝光日志流）均为教学用假组件，不涉及任何真实 RPC / 存储。31"""3233import math34import random3536# ---------------------------------------------------------------- 业务常量37# 正文表「不同业务价值的兑换关系示例」：以「分钟时长」为统一价值单位，1 分钟 = 0.05 元38MINUTE_VALUE_RMB = 0.0539EXCHANGE_RATE = {40    "video": 1.0,    # 预计观看 x 分钟 → x 分钟时长价值41    "ad": 20.0,      # 广告收入 1 元 → 20 分钟42    "live": 10.0,    # 直播打赏 1 元 → 10 分钟43    "ecom": 2.0,     # 电商 GMV 1 元 → 2 分钟44}454647# ============================================================ [MOCK] 业务侧48class BusinessRecSys:49    """[MOCK] 单个业务的推荐系统：内部已完成召回/排序，对外只吐出 Top-N 候选。"""5051    def __init__(self, biz: str, n_items: int, raw_scale: float, bias: float):52        self.biz = biz53        self.n_items = n_items54        self.raw_scale = raw_scale   # 业务原始分量纲（广告 eCPM 可能上千，pCTR 只在 [0,1]）55        self.bias = bias             # 模型系统性高估/低估系数，>1 表示高估5657    def serve(self) -> list[dict]:58        out = []59        for i in range(self.n_items):60            true_p = random.betavariate(2.0, 5.0)              # 真实后验概率61            pred_p = min(0.999, true_p * self.bias + random.gauss(0, 0.02))62            pred_p = max(1e-4, pred_p)63            out.append({64                "id": f"{self.biz[0].upper()}{i + 1}",65                "biz": self.biz,66                "true_p": true_p,67                "pxtr": pred_p,                                # 业务模型原始预估68                "raw_score": pred_p * self.raw_scale,          # 业务原始竞价分（量纲各异）69            })70        out.sort(key=lambda d: -d["raw_score"])71        return out727374class Router:75    """[MOCK] 推荐 Router：并发请求各业务推荐系统，把多路结果透传给混排服务。"""7677    def __init__(self, systems: list[BusinessRecSys]):78        self.systems = systems7980    def fanout(self) -> dict[str, list[dict]]:81        return {s.biz: s.serve() for s in self.systems}828384class ExposureLogStream:85    """[MOCK] 曝光日志流：滑动窗口统计各业务曝光占比，供 Load 调控读取。"""8687    def __init__(self, window: int = 200):88        self.window, self.buf = window, []8990    def report(self, seq: list[dict]) -> None:91        for it in seq:92            self.buf.append(it["biz"])93        self.buf = self.buf[-self.window:]9495    def load_of(self, biz: str) -> float:96        return 0.0 if not self.buf else sum(b == biz for b in self.buf) / len(self.buf)979899# ===================================================== 阶段 1：混排前处理100def pava_calibrate(pairs: list[tuple[float, float]], n_bucket: int = 10):101    """保序回归（PAVA）：按 pxtr 分桶统计真实后验，再合并违反单调的相邻桶。"""102    pairs = sorted(pairs, key=lambda t: t[0])103    size = max(1, len(pairs) // n_bucket)104    blocks = []105    for s in range(0, len(pairs), size):106        chunk = pairs[s:s + size]107        if not chunk:108            continue109        blocks.append({110            "lo": chunk[0][0], "hi": chunk[-1][0],111            "n": len(chunk), "r": sum(y for _, y in chunk) / len(chunk),112        })113    # PAVA：相邻违序则加权合并114    merged: list[dict] = []115    for b in blocks:116        merged.append(b)117        while len(merged) >= 2 and merged[-2]["r"] > merged[-1]["r"]:118            b2, b1 = merged.pop(), merged.pop()119            n = b1["n"] + b2["n"]120            merged.append({"lo": b1["lo"], "hi": b2["hi"], "n": n,121                           "r": (b1["n"] * b1["r"] + b2["n"] * b2["r"]) / n})122    return merged123124125def apply_calibration(table: list[dict], p: float) -> float:126    """查表 + 区间线性插值（SIR 的平滑思想），避免桶边界跳变。"""127    mids = [(b["lo"] + b["hi"]) / 2 for b in table]128    if p <= mids[0]:129        return table[0]["r"]130    if p >= mids[-1]:131        return table[-1]["r"]132    for k in range(len(mids) - 1):133        if mids[k] <= p <= mids[k + 1]:134            w = (p - mids[k]) / max(1e-12, mids[k + 1] - mids[k])135            return table[k]["r"] + w * (table[k + 1]["r"] - table[k]["r"])136    return table[-1]["r"]137138139def minmax(values: list[float]) -> list[float]:140    lo, hi = min(values), max(values)141    return [0.5] * len(values) if hi - lo < 1e-12 else [(v - lo) / (hi - lo) for v in values]142143144def pre_process(lists: dict[str, list[dict]], logs: dict[str, list[tuple[float, float]]]):145    """校准 + 归一化 + 统一价值折算，输出可跨业务比较的 Value(item)。"""146    report = {}147    for biz, items in lists.items():148        table = pava_calibrate(logs[biz])149        for it in items:150            it["calibrated"] = apply_calibration(table, it["pxtr"])151        before = sum(i["pxtr"] for i in items) / sum(i["true_p"] for i in items)152        after = sum(i["calibrated"] for i in items) / sum(i["true_p"] for i in items)153        norm = minmax([i["calibrated"] for i in items])154        for it, z in zip(items, norm):155            it["norm"] = z156            # 统一价值：归一化分 × 业务侧的原始业务量 × 兑换系数157            it["value"] = EXCHANGE_RATE[biz] * it["calibrated"] * BIZ_UNIT[biz]158        report[biz] = (before, after, len(table))159    return report160161162# 各业务「一次曝光对应的业务量」：短视频=预计分钟数；广告=eCPM 元；直播=打赏元；电商=GMV 元163BIZ_UNIT = {"video": 6.0, "ad": 0.5, "live": 1.2, "ecom": 4.0}164165166# ===================================================== 阶段 2：序列生成167def gen_sequences(video: list[dict], ads: list[dict], lives: list[dict],168                  max_ad: int, ad_gap: int, live_first_allowed: bool,169                  prev_tail_is_ad: bool) -> list[list[dict]]:170    """在短视频骨架上枚举广告/直播插入方案，同时执行约束剪枝。"""171    seqs = []172173    def ok(seq: list[dict]) -> bool:174        if prev_tail_is_ad and seq[0]["biz"] == "ad":      # 跨刷次 Gap 约束175            return False176        if not live_first_allowed and seq[0]["biz"] == "live":177            return False178        last_ad = -10179        for i, it in enumerate(seq):180            if it["biz"] == "ad":181                if i - last_ad <= ad_gap:                   # 刷次内 Gap 约束182                    return False183                last_ad = i184        return True185186    for n_ad in range(max_ad + 1):187        for n_live in (0, 1) if lives else (0,):188            inserts = ads[:n_ad] + lives[:n_live]189            total = len(video) + len(inserts)190            for mask in _combinations(range(total), len(inserts)):191                seq, vi, ii = [], 0, 0192                slots = set(mask)193                for pos in range(total):194                    if pos in slots:195                        seq.append(inserts[ii]); ii += 1196                    else:197                        seq.append(video[vi]); vi += 1198                if ok(seq):199                    seqs.append(seq)200    return seqs201202203def _combinations(pool, k):204    pool = list(pool)205    if k == 0:206        yield ()207        return208    for i in range(len(pool) - k + 1):209        for rest in _combinations(pool[i + 1:], k - 1):210            yield (pool[i],) + rest211212213# ===================================================== 阶段 3：序列排序214def seq_score(seq: list[dict], lam: float, mock_value: float, target_len: int) -> float:215    """Score_seq = Σ γ_i · Value(item_i)，不足长度用 Mock Item 补齐。"""216    s = 0.0217    for i in range(target_len):218        gamma = math.exp(-lam * i)219        s += gamma * (seq[i]["value"] if i < len(seq) else mock_value)220    return s221222223# ===================================================== 阶段 4：混排后处理224class LoadController:225    """PID Load 调控：u_t = Kp·e_t + Ki·Σe + Kd·(e_t - e_{t-1})，score_final = score_raw + u_t。"""226227    def __init__(self, target: float, kp=1.2, ki=0.25, kd=0.3,228                 max_integral=1.0, max_adjust=0.4):229        self.target, self.kp, self.ki, self.kd = target, kp, ki, kd230        self.integral, self.last_err = 0.0, 0.0231        self.max_integral, self.max_adjust = max_integral, max_adjust232233    def step(self, current: float) -> float:234        e = self.target - current235        self.integral = max(-self.max_integral, min(self.max_integral, self.integral + e))236        u = self.kp * e + self.ki * self.integral + self.kd * (e - self.last_err)237        self.last_err = e238        return max(-self.max_adjust, min(self.max_adjust, u))239240241def post_process(seq: list[dict], ad_adjust: float, force_item: dict | None,242                 force_pos: int, demote_tags: set[str], quota_biz: str,243                 quota_min: int) -> list[dict]:244    """后处理：Load 调节量修分 → 降权 → 保量兜底 → 运营强插（优先级最高）。"""245    out = [dict(it) for it in seq]246    for it in out:247        if it["biz"] == "ad":248            it["value"] += ad_adjust * 5.0            # Load 调控作用于排序分249        if it.get("tag") in demote_tags:250            it["value"] *= 0.5                        # 降权策略251    # 保量：目标业务曝光数不足则把末位替换为该业务兜底内容252    have = sum(1 for it in out if it["biz"] == quota_biz)253    if have < quota_min:254        out[-1] = {"id": "QUOTA-1", "biz": quota_biz, "value": 1.0, "tag": "quota"}255    # 强插：最后阶段执行，优先级高于模型排序256    if force_item is not None:257        out.insert(min(force_pos, len(out)), force_item)258    return out259260261# ===================================================== 演示262def main() -> None:263    random.seed(42)264    print("=" * 74)265    print("混排模块端到端链路演示（§8.1 图 8.2 的四个环节）")266    print("=" * 74)267268    router = Router([269        BusinessRecSys("video", 6, 1.0, 1.05),270        BusinessRecSys("ad", 3, 1800.0, 1.45),     # 广告系统明显高估271        BusinessRecSys("live", 2, 30.0, 0.80),     # 直播系统低估272        BusinessRecSys("ecom", 2, 120.0, 1.15),273    ])274    lists = router.fanout()275276    print("\n【阶段 0】Router 透传的多业务候选（raw_score 量纲完全不可比）")277    for biz, items in lists.items():278        head = ", ".join(f"{i['id']}:{i['raw_score']:.2f}" for i in items[:3])279        print(f"  {biz:<6} n={len(items)}  raw_score 前三 = {head}")280281    # 离线日志（用于拟合校准表）282    logs = {}283    for biz, sysm in zip(lists, router.systems):284        rows = []285        for _ in range(4000):286            t = random.betavariate(2.0, 5.0)287            p = max(1e-4, min(0.999, t * sysm.bias + random.gauss(0, 0.02)))288            rows.append((p, 1.0 if random.random() < t else 0.0))289        logs[biz] = rows290291    rep = pre_process(lists, logs)292    print("\n【阶段 1】混排前处理：IR 校准 + Min-Max 归一化 + 统一价值折算")293    print(f"  {'业务':<8}{'PCOC(校准前)':>14}{'PCOC(校准后)':>14}{'IR桶数':>8}")294    for biz, (b, a, nb) in rep.items():295        print(f"  {biz:<8}{b:>14.4f}{a:>14.4f}{nb:>8d}")296    print("  统一价值分 Value(item)（已折算为分钟时长价值）:")297    for biz, items in lists.items():298        print(f"    {biz:<6} " + ", ".join(f"{i['id']}={i['value']:.2f}" for i in items))299300    video, ads = lists["video"], lists["ad"]301    lives = lists["live"]302    seqs = gen_sequences(video, ads, lives, max_ad=3, ad_gap=3,303                         live_first_allowed=True, prev_tail_is_ad=True)304    print(f"\n【阶段 2】序列生成与剪枝：合法候选序列 {len(seqs)} 条"305          f"（约束：刷次内广告间隔≥3，上一刷尾部是广告故首位禁广告）")306307    all_vals = [i["value"] for items in lists.values() for i in items]308    mock_v = sum(all_vals) / len(all_vals)309    target_len = max(len(s) for s in seqs)310    ranked = sorted(seqs, key=lambda s: -seq_score(s, 0.2, mock_v, target_len))311    best = ranked[0]312    print(f"\n【阶段 3】序列排序：λ=0.2 指数位置衰减，Mock Item 价值={mock_v:.3f}，"313          f"对齐长度={target_len}")314    for rank, s in enumerate(ranked[:3], 1):315        print(f"  Top{rank} score={seq_score(s, 0.2, mock_v, target_len):7.3f}  "316              + " → ".join(f"{i['id']}({i['biz'][:3]})" for i in s))317318    stream = ExposureLogStream(window=120)319    for s in ranked[:20]:320        stream.report(s)321    ctrl = LoadController(target=0.12)322    cur = stream.load_of("ad")323    u = ctrl.step(cur)324    print(f"\n【阶段 4】混排后处理与流量调控")325    print(f"  滑动窗口广告 Load = {cur:.4f}，目标 = 0.1200，PID 调节量 u_t = {u:+.4f}")326    forced = {"id": "OP-LIVE", "biz": "live", "value": 9.9, "tag": "operation"}327    final = post_process(best, u, forced, force_pos=2,328                         demote_tags={"aigc"}, quota_biz="ecom", quota_min=1)329    print("  最终曝光序列：")330    for i, it in enumerate(final, 1):331        print(f"    pos{i:<2} {it['id']:<8} biz={it['biz']:<6} value={it['value']:.3f}"332              + ("   ← 运营强插" if it.get("tag") == "operation" else "")333              + ("   ← 保量兜底" if it.get("tag") == "quota" else ""))334    print(f"  序列长度 {len(best)} → {len(final)}，"335          f"广告占比 {sum(i['biz'] == 'ad' for i in final) / len(final):.3f}")336337338if __name__ == "__main__":339    main()