quota_control.py

1"""2粗排 Quota 调控:固定配额 / 动态配额 / 闭环控制(第 5 章 §5.6.1 粗排Quota调控)34对应正文概念5------------6正文先用表 prerank_no_quota_example 说明:若完全按 TopK(score) 筛选,候选集可能7全部来自同一个池(5 个游戏主播 0.98/0.96/0.95/0.94/0.93),短期指标最优但带来内容同质化、8新主播无曝光、生态失衡等长期问题。于是引入 Quota 机制,表 prerank_quota_config_example9给出的配置是:游戏直播池 40%、娱乐直播池 30%、电商直播池 20%、新主播池 10%,10对应 100 个输出候选时:1112    Quota_game = 40, Quota_entertainment = 30, Quota_ecommerce = 20, Quota_new = 101314正文 code_prerank_quota 伪代码的逻辑是:多路召回 -> 粗排模型预测 pctr/pcvr/pltr ->15融合 score = 0.5 pctr + 0.3 pcvr + 0.2 pltr -> 按业务规则切池 -> 池内按 score 排序 ->16各池取 TopQuota_i -> CandidateSet = ∪_i TopK_i。1718正文接着指出固定 Quota 灵活性差:对重度游戏用户 P(Game|User) = 0.9 时仍给 40% 不合理,19因此引入动态 Quota:2021    Quota_i = f(User, Context, BusinessGoal)2223正文举例可调成 Quota_game = 50%、Quota_ecommerce = 10%;更进一步可用强化学习/在线优化24学习策略 π(s) -> Quota,根据实时业务指标动态调整流量结构。2526本实现做了什么27--------------281. 完整复刻 code_prerank_quota:[MOCK] retrieve() 多路召回 + 三个 [MOCK] pxtr 模型29   + 0.5/0.3/0.2 融合 + 切池 + 池内排序 + 固定 Quota 取 TopK,并与无约束 TopK(score)30   对比池分布、分数损失、生态指标。312. 动态 Quota:Quota_i = f(User, Context, BusinessGoal) 的一个可解释实现 ——32   用户池偏好 P(pool|user)、业务目标权重与生态保底三者做凸组合后按最大余数法取整,33   打印重度游戏用户 / 电商意图用户 / 新用户三种画像下的配额与命中分数。343. 闭环控制:把 π(s) -> Quota 写成一个带积分项的离散反馈控制律35       e_t = target_share - realized_share_t36       q_{t+1} = clip(q_t + K_p e_t + K_i Σ_{τ≤t} e_τ, q_min, q_max)37   然后归一化成整数配额。[MOCK] 环境给出「下游精排/用户对各池的实际消费倾向」,38   使开环固定配额无法达到曝光目标,闭环控制在若干轮内把误差压到阈值内。39   打印逐轮(轮次 × 池)的配额、实际曝光占比与误差,以及收敛轮次。4041关键公式42--------43    score = 0.5 p_ctr + 0.3 p_cvr + 0.2 p_ltr44    CandidateSet = ∪_i TopK_i,  |TopK_i| = Quota_i45    Quota_i = f(User, Context, BusinessGoal)46    π(s) -> Quota:  q_{t+1} = clip(q_t + K_p e_t + K_i Σ e_τ, q_min, q_max)47"""4849from __future__ import annotations5051import math52import random53from dataclasses import dataclass, field5455OUTPUT_SIZE = 100                       # 正文:粗排最终输出 100 个候选56FIXED_QUOTA = {"game": 40, "entertainment": 30, "ecommerce": 20, "new": 10}57FUSION_W = {"pctr": 0.5, "pcvr": 0.3, "pltr": 0.2}   # 正文 code_prerank_quota58POOLS = ("game", "entertainment", "ecommerce", "new")596061@dataclass62class Item:63    """一个候选主播/直播间。"""64    iid: int65    category: str           # game / entertainment / ecommerce66    is_new_anchor: bool67    pctr: float = 0.068    pcvr: float = 0.069    pltr: float = 0.070    score: float = 0.0717273@dataclass74class MockModels:75    """[MOCK] 三个粗排 pxtr 模型:按类目/新主播身份给出带噪声的预测值,不做真实推理。"""76    rng: random.Random77    base: dict[str, tuple[float, float, float]] = field(default_factory=lambda: {78        # category -> (pctr 均值, pcvr 均值, pltr 均值)79        "game": (0.085, 0.020, 0.055),80        "entertainment": (0.060, 0.012, 0.070),81        "ecommerce": (0.045, 0.060, 0.020),82    })8384    def predict(self, it: Item) -> None:85        mc, mv, ml = self.base[it.category]86        penalty = 0.72 if it.is_new_anchor else 1.0     # 新主播缺少后验,预测分天然偏低87        it.pctr = max(1e-4, self.rng.gauss(mc * penalty, mc * 0.35))88        it.pcvr = max(1e-5, self.rng.gauss(mv * penalty, mv * 0.40))89        it.pltr = max(1e-4, self.rng.gauss(ml * penalty, ml * 0.35))909192def retrieve(n: int, rng: random.Random) -> list[Item]:93    """[MOCK] 多路召回:按类目/新主播比例生成候选,不访问任何真实索引。"""94    out = []95    for i in range(n):96        r = rng.random()97        cat = "game" if r < 0.55 else ("entertainment" if r < 0.82 else "ecommerce")98        out.append(Item(iid=i, category=cat, is_new_anchor=rng.random() < 0.18))99    return out100101102def pool_of(it: Item) -> str:103    """正文切池规则:new_anchor_pool 由 is_new_anchor 定义,与类目池并列。"""104    return "new" if it.is_new_anchor else it.category105106107def fuse(items: list[Item]) -> None:108    for it in items:109        it.score = (FUSION_W["pctr"] * it.pctr + FUSION_W["pcvr"] * it.pcvr110                    + FUSION_W["pltr"] * it.pltr)111112113def split_pools(items: list[Item]) -> dict[str, list[Item]]:114    """按业务规则切池,并在池内按 score 降序排序(正文伪代码第二段)。"""115    pools: dict[str, list[Item]] = {p: [] for p in POOLS}116    for it in items:117        pools[pool_of(it)].append(it)118    for p in pools:119        pools[p].sort(key=lambda x: -x.score)120    return pools121122123def apply_quota(pools: dict[str, list[Item]], quota: dict[str, int]) -> list[Item]:124    """CandidateSet = ∪_i TopK_i。"""125    result: list[Item] = []126    for p in POOLS:127        result.extend(pools[p][: quota.get(p, 0)])128    return result129130131def unconstrained_topk(items: list[Item], k: int) -> list[Item]:132    return sorted(items, key=lambda x: -x.score)[:k]133134135def share(sel: list[Item]) -> dict[str, float]:136    n = max(1, len(sel))137    cnt = {p: 0 for p in POOLS}138    for it in sel:139        cnt[pool_of(it)] += 1140    return {p: cnt[p] / n for p in POOLS}141142143def entropy(sh: dict[str, float]) -> float:144    h = -sum(v * math.log(v) for v in sh.values() if v > 0)145    return h / math.log(len(sh))146147148def largest_remainder(weights: dict[str, float], total: int) -> dict[str, int]:149    """最大余数法:把连续份额转成整数配额,保证 Σ Quota_i = total。"""150    raw = {p: weights[p] * total for p in weights}151    base = {p: int(raw[p]) for p in raw}152    rest = total - sum(base.values())153    for p, _ in sorted(raw.items(), key=lambda kv: -(kv[1] - int(kv[1])))[:rest]:154        base[p] += 1155    return base156157158def dynamic_quota(user_pref: dict[str, float], biz_goal: dict[str, float],159                  floor: dict[str, float], total: int,160                  w_user: float = 0.65, w_biz: float = 0.20,161                  w_floor: float = 0.15) -> dict[str, int]:162    """Quota_i = f(User, Context, BusinessGoal):三路份额的凸组合 + 最大余数取整。"""163    mix = {p: w_user * user_pref.get(p, 0.0) + w_biz * biz_goal.get(p, 0.0)164              + w_floor * floor.get(p, 0.0) for p in POOLS}165    tot = sum(mix.values())166    return largest_remainder({p: v / tot for p, v in mix.items()}, total)167168169class QuotaController:170    """π(s) -> Quota 的 PI 反馈控制律(离散时间,带积分项与饱和限幅)。"""171172    def __init__(self, target: dict[str, float], kp: float = 0.45, ki: float = 0.30,173                 q_min: float = 0.03, q_max: float = 0.70):174        self.target = dict(target)175        self.kp, self.ki = kp, ki176        self.q_min, self.q_max = q_min, q_max177        self.q = dict(target)                        # 初始配额 = 目标曝光占比178        self.integral = {p: 0.0 for p in target}179180    def act(self) -> dict[str, int]:181        tot = sum(self.q.values())182        return largest_remainder({p: v / tot for p, v in self.q.items()}, OUTPUT_SIZE)183184    def update(self, realized: dict[str, float]) -> dict[str, float]:185        err = {p: self.target[p] - realized.get(p, 0.0) for p in self.target}186        for p in self.q:187            self.integral[p] += err[p]188            nq = self.q[p] + self.kp * err[p] + self.ki * self.integral[p]189            self.q[p] = min(self.q_max, max(self.q_min, nq))190        return err191192193def mock_downstream(quota: dict[str, int], pass_rate: dict[str, float],194                    rng: random.Random, requests: int = 200) -> dict[str, float]:195    """[MOCK] 下游环境:一轮内有 `requests` 次请求,各池候选以不同通过率进入最终曝光,196    导致开环配额与目标曝光结构失真(请求数越大,统计噪声越小)。"""197    exposed = {p: 0.0 for p in quota}198    for p, q in quota.items():199        trials = q * requests200        got = sum(1 for _ in range(trials) if rng.random() < pass_rate[p])201        exposed[p] = float(got)202    tot = max(1.0, sum(exposed.values()))203    return {p: v / tot for p, v in exposed.items()}204205206if __name__ == "__main__":207    random.seed(42)208    rng = random.Random(42)209210    # ---- 0. 正文表 prerank_no_quota_example ----211    print("=" * 100)212    print("正文表 prerank_no_quota_example:无 Quota 约束时 Top5 全部来自游戏池")213    print("=" * 100)214    demo = [("A", "游戏", 0.98), ("B", "游戏", 0.96), ("C", "游戏", 0.95),215            ("D", "游戏", 0.94), ("E", "游戏", 0.93)]216    print(f"{'主播ID':>8}{'类别':>8}{'Score':>10}")217    for a, b, c in demo:218        print(f"{a:>8}{b:>8}{c:>10.2f}")219    print("短期业务指标最优,但内容同质化、新主播零曝光 -- 正文据此引入 Quota 机制。")220221    # ---- 1. 复刻 code_prerank_quota:固定 Quota vs 无约束 TopK ----222    print("\n" + "=" * 100)223    print("复刻 code_prerank_quota:召回 5000 -> 粗排打分 -> 0.5/0.3/0.2 融合 -> 固定 Quota")224    print("=" * 100)225    cands = retrieve(5000, rng)226    models = MockModels(rng)227    for it in cands:228        models.predict(it)229    fuse(cands)230    pools = split_pools(cands)231    print(f"配额配置 {FIXED_QUOTA},Σ Quota_i = {sum(FIXED_QUOTA.values())} = 输出候选数")232    print(f"{'池':>16}{'池内候选数':>12}{'池内最高 score':>16}{'池内第 Quota 名 score':>22}")233    for p in POOLS:234        q = FIXED_QUOTA[p]235        top = pools[p][0].score if pools[p] else 0.0236        cut = pools[p][q - 1].score if len(pools[p]) >= q else float("nan")237        print(f"{p:>16}{len(pools[p]):>12}{top:>16.5f}{cut:>22.5f}")238239    quota_set = apply_quota(pools, FIXED_QUOTA)240    free_set = unconstrained_topk(cands, OUTPUT_SIZE)241    print(f"\n{'策略':>18}{'候选数':>8}" + "".join(f"{p+'占比':>14}" for p in POOLS)242          + f"{'平均 score':>12}{'池熵':>8}{'新主播数':>10}")243    for name, sel in (("无约束 TopK(score)", free_set), ("固定 Quota ∪TopK_i", quota_set)):244        sh = share(sel)245        avg = sum(x.score for x in sel) / len(sel)246        newn = sum(1 for x in sel if x.is_new_anchor)247        print(f"{name:>18}{len(sel):>8}" + "".join(f"{sh[p]:>14.3f}" for p in POOLS)248              + f"{avg:>12.5f}{entropy(sh):>8.4f}{newn:>10}")249    a_free = sum(x.score for x in free_set) / len(free_set)250    a_quota = sum(x.score for x in quota_set) / len(quota_set)251    print(f"\nQuota 让平均融合分从 {a_free:.5f} 降到 {a_quota:.5f}"252          f"(相对损失 {(a_free - a_quota) / a_free:.2%}),"253          f"换来池熵 {entropy(share(free_set)):.4f} -> {entropy(share(quota_set)):.4f}"254          f" 与新主播曝光 {sum(1 for x in free_set if x.is_new_anchor)} -> "255          f"{sum(1 for x in quota_set if x.is_new_anchor)}。")256257    # ---- 2. 动态 Quota ----258    print("\n" + "=" * 100)259    print("动态 Quota:Quota_i = f(User, Context, BusinessGoal)(凸组合 + 最大余数取整)")260    print("=" * 100)261    biz = {"game": 0.20, "entertainment": 0.20, "ecommerce": 0.50, "new": 0.10}262    floor = {"game": 0.25, "entertainment": 0.25, "ecommerce": 0.25, "new": 0.25}263    profiles = {264        "重度游戏用户 P(Game|U)=0.9": {"game": 0.90, "entertainment": 0.06,265                                 "ecommerce": 0.03, "new": 0.01},266        "电商意图用户": {"game": 0.15, "entertainment": 0.15,267                    "ecommerce": 0.65, "new": 0.05},268        "冷启新用户(兴趣未知)": {"game": 0.30, "entertainment": 0.30,269                          "ecommerce": 0.20, "new": 0.20},270    }271    print(f"{'用户画像':>24}" + "".join(f"{p:>14}" for p in POOLS)272          + f"{'平均 score':>12}{'池熵':>8}")273    print(f"{'固定 Quota(基线)':>24}"274          + "".join(f"{FIXED_QUOTA[p]:>14}" for p in POOLS)275          + f"{a_quota:>12.5f}{entropy(share(quota_set)):>8.4f}")276    for name, pref in profiles.items():277        dq = dynamic_quota(pref, biz, floor, OUTPUT_SIZE)278        sel = apply_quota(pools, dq)279        sh = share(sel)280        avg = sum(x.score for x in sel) / len(sel)281        print(f"{name:>24}" + "".join(f"{dq[p]:>14}" for p in POOLS)282              + f"{avg:>12.5f}{entropy(sh):>8.4f}")283    dq_game = dynamic_quota(profiles["重度游戏用户 P(Game|U)=0.9"], biz, floor, OUTPUT_SIZE)284    print(f"\n重度游戏用户的游戏配额由固定 40 提升到 {dq_game['game']},"285          f"电商配额由 20 降到 {dq_game['ecommerce']} -- 与正文「Quota_game=50%、"286          f"Quota_ecommerce=10%」的调控方向一致;")287    print("生态保底项 floor 保证了即使用户兴趣极度集中,各池仍有最低配额,不会被完全挤出。")288289    # ---- 3. 闭环控制 π(s) -> Quota ----290    print("\n" + "=" * 100)291    print("闭环 Quota 调控 π(s)->Quota:PI 控制律把实际曝光占比拉回目标")292    print("=" * 100)293    target = {"game": 0.40, "entertainment": 0.30, "ecommerce": 0.20, "new": 0.10}294    # [MOCK] 下游通过率:电商与新主播在精排/用户侧更容易被淘汰,开环配额必然偏离目标295    pass_rate = {"game": 0.95, "entertainment": 0.80, "ecommerce": 0.45, "new": 0.35}296    print(f"目标曝光占比 {target}")297    print(f"[MOCK] 下游通过率 {pass_rate}\n")298299    print("开环(固定配额,不反馈):")300    rng_open = random.Random(7)301    open_err = []302    for t in range(1, 7):303        real = mock_downstream(FIXED_QUOTA, pass_rate, rng_open)304        mae = sum(abs(target[p] - real[p]) for p in POOLS) / len(POOLS)305        open_err.append(mae)306        print(f"  轮 {t}: 实际占比 " + " ".join(f"{p}={real[p]:.3f}" for p in POOLS)307              + f" | MAE={mae:.4f}")308309    print("\n闭环(PI 控制,K_p=0.45, K_i=0.30):")310    ctrl = QuotaController(target)311    rng_cl = random.Random(7)312    conv = None313    print(f"{'轮':>4}" + "".join(f"{p+' q/占比':>18}" for p in POOLS) + f"{'MAE':>9}")314    for t in range(1, 13):315        q = ctrl.act()316        real = mock_downstream(q, pass_rate, rng_cl)317        mae = sum(abs(target[p] - real[p]) for p in POOLS) / len(POOLS)318        if conv is None and mae <= 0.02:319            conv = t320        print(f"{t:>4}" + "".join(f"{q[p]:>9}/{real[p]:>8.3f}" for p in POOLS)321              + f"{mae:>9.4f}")322        ctrl.update(real)323    print(f"\n开环 MAE 均值 {sum(open_err)/len(open_err):.4f},始终无法收敛(电商/新主播常年欠曝光);")324    print(f"闭环首次把 MAE 压到 ≤0.02 的轮次 = {conv};稳态配额 "325          + " ".join(f"{p}={ctrl.act()[p]}" for p in POOLS))326    print("控制律通过放大高损耗池的配额来补偿下游损失,这正是 π(s)->Quota 在线优化的最小形态。")