7.3 分布漂移检测：PSI与分桶监控

7.3 分布漂移检测：PSI与分桶监控

本节摘要：第 1.2 节的三重风险里，输入漂移最隐蔽：代码一行没改、模型一个字没动，用户群的构成变了，质量就悄悄滑走——而业务指标（客诉率、负反馈率）是滞后信号，等它报警往往已过数周。对冲办法是监控输入与中间量的分布：置信度、场景标签、输入长度、分桶命中率——它们是漂移的先导指标。度量工具用 PSI（Population Stability Index，群体稳定性指数）：PSI = Σ(实际占比 − 预期占比) × ln(实际占比 / 预期占比)，把"本周分布"与"基线分布"在分桶后逐桶对比，压缩成一个数。风控业通行口径（社区）：<0.1 稳定、0.1~0.25 观察、>0.25 显著漂移——立即触发抽样人工与回归集复跑。工程要点有三：数值型变量用等频分桶（基线分位数定桶边界，避免空桶与长尾失衡）；零桶用小常数平滑（ln(0) 会炸）；漂移检测要多维同时看（单看均值会被"左右互搏"的分布变形骗过）。本节附完整 PSI 实现与周度体检报告脚本（MOCK 可跑），并给出每周的固定动作清单。

学习目标

阅读完本节，你应当能够：

1. 手算一个两桶例子的 PSI 并按三级口径判读。
2. 实现等频分桶、平滑与多维 PSI 矩阵。
3. 把"周一跑、周四议、触发即动作"的体检节奏落地。
4. 说出 PSI 的两个盲区与补救办法。

一、为什么是分布，为什么是 PSI

监控对象选输入与中间量的分布而非业务指标，理由是时滞：

PSI 的直觉：对每个桶算"实际占比相对预期占比的倍数"取对数，再乘占比差加权——占比不变的桶贡献为 0；质量从 A 桶大规模迁往 B 桶时，两桶各贡献一大块，加总放大。手算例（示意）：基线 {高置信 80%, 低置信 20%}，本周 {高置信 60%, 低置信 40%}：

PSI = (0.6−0.8)×ln(0.6/0.8) + (0.4−0.2)×ln(0.4/0.2)
    = (−0.2)×(−0.288) + (0.2)×(0.693) ≈ 0.196   → 落入"观察"档（0.1~0.25）
​

二、完整脚本：PSI 矩阵 + 周度体检报告（MOCK 可跑）

# psi_monitor.py（纯标准库，可直接运行）
"""PSI 漂移检测：等频分桶 + 平滑 + 多维周报。基线来自上线时的稳定周期。"""
import math
import random

def equal_freq_edges(baseline: list[float], bins: int = 10) -> list[float]:
    """等频分桶：边界取基线的分位数，保证基线每桶样本量近似相等。"""
    s = sorted(baseline)
    return [s[min(len(s) - 1, int(len(s) * k / bins))] for k in range(1, bins)]

def bucketize(values: list[float], edges: list[float]) -> list[int]:
    """按边界把值分桶（返回桶号 0..bins）。"""
    return [sum(v > e for e in edges) for v in values]

def psi(expected: list[float], actual: list[float], bins: int = 10,
        smooth: float = 1e-4) -> tuple[float, list[tuple[int, float, float]]]:
    """返回 (PSI, [(桶号, 基线占比, 实际占比)])。数值型输入自动等频分桶。"""
    edges = equal_freq_edges(expected, bins)
    e_bins = bucketize(expected, edges)
    a_bins = bucketize(actual, edges)
    n_e, n_a = len(e_bins), len(a_bins)
    detail, total = [], 0.0
    for b in range(bins + 1):
        pe = max((e_bins.count(b) / n_e), smooth)      # 平滑：零桶防 ln 爆炸
        pa = max((a_bins.count(b) / n_a), smooth)
        total += (pa - pe) * math.log(pa / pe)
        detail.append((b, e_bins.count(b) / n_e, a_bins.count(b) / n_a))
    return total, detail

def classify(value: float) -> str:
    """三级口径（风控通行，社区经验）。"""
    if value < 0.10:
        return "稳定"
    return "观察" if value < 0.25 else "★漂移★"

def psi_for_categorical(base: dict, cur: dict, smooth: float = 1e-4) -> float:
    """分类变量（如场景标签）的 PSI：直接按类别占比算。"""
    keys = set(base) | set(cur)
    nb, nc = sum(base.values()), sum(cur.values())
    return sum((cur.get(k, 0) / nc - base.get(k, 0) / nb)
               * math.log(max(cur.get(k, 0) / nc, smooth)
                          / max(base.get(k, 0) / nb, smooth)) for k in keys)

if __name__ == "__main__":
    rng = random.Random(7)
    # 基线：上线月（示意）。三个被监控量：置信度 / 输入长度 / 场景占比
    base_conf = [rng.betavariate(9, 2) for _ in range(5000)]
    base_len = [rng.gauss(180, 60) for _ in range(5000)]
    base_scn = {"billing": 4000, "technical": 800, "security": 200}
    weeks = {
        "W1": (lambda: [rng.betavariate(9, 2) for _ in range(2000)],
               lambda: [rng.gauss(182, 61) for _ in range(2000)],
               {"billing": 1560, "technical": 370, "security": 70}),
        "W2": (lambda: [rng.betavariate(8.2, 2.5) for _ in range(2000)], # 置信轻度下移
               lambda: [rng.gauss(179, 58) for _ in range(2000)],
               {"billing": 1490, "technical": 440, "security": 70}),
        "W3": (lambda: [rng.betavariate(4, 4) for _ in range(2000)],   # 大幅下移
               lambda: [rng.gauss(240, 90) for _ in range(2000)],     # 输入变长
               {"billing": 900, "technical": 1000, "security": 100}), # 场景迁移
    }
    print("周度体检：<0.10 稳定 | 0.10~0.25 观察 | >0.25 ★漂移★")
    for name, (conf_f, len_f, scn) in weeks.items():
        v1, _ = psi(base_conf, conf_f())
        v2, _ = psi(base_len, len_f())
        v3 = psi_for_categorical(base_scn, scn)
        print(f"{name}  置信度 PSI={v1:.2f} [{classify(v1)}]"
              f"  输入长度 PSI={v2:.2f} [{classify(v2)}]"
              f"  场景占比 PSI={v3:.2f} [{classify(v3)}]")
​

运行输出（实测，本机 Python 3.12）：W1 三项全部"稳定"（各 PSI < 0.1）；W2 置信度 PSI ≈ 0.18 进入"观察"（长度、场景仍稳定）；W3 三项齐报"★漂移★"（置信度 PSI > 3、长度 0.6+、场景 0.5+）——一条随周次逐步升级的漂移时间线，正是输入漂移"慢刀子"的典型形态。

三、分桶与平滑的工程细节

1. 等频优于等宽：等频分桶让基线每桶占比 ≈ 1/bins，PSI 对小幅移动更敏感且无空桶；等宽桶在长尾变量（长度、成本）上会把 90% 样本塞进一两个桶，PSI 变钝；
2. 桶数 10 左右（社区惯例）：样本少于 1000 时降到 5 桶，否则每桶太稀，PSI 抖动像噪声；
3. 平滑保平安：max(占比, 1e-4) 一行防御 ln(0)；新类别出现（场景表里多了个键）本身就是漂移信号，别把它平滑没了——分类 PSI 前先 diff 键集合；
4. 基线要可辩护：选"系统稳定、分布正常"的周期（如上线后首月），并随重大版本更新而重立（第 3.4 节版本化思想）——拿异常月当基线，之后周周报警。

四、周度体检节奏

周一上午：跑 PSI 矩阵（脚本定时任务，数据来自 6.2 的 trace 或日志抽样）
   ├─ 全绿（<0.10）           → 归档，无动作
   ├─ 观察（0.10~0.25）        → 周四评审会过一遍：抽样 50 条人工快看
   │                              连续两周同向观察 → 升级处理
   └─ ★漂移★（>0.25）          → 当日动作：
         ① 分桶下钻：哪个桶在涨/跌（脚本 detail 列表）
         ② 抽样 100 条人工评估质量是否受损
         ③ 回归集在新分布上复跑（第 3 章考卷验证）
         ④ 受损 → 触发第 8.3 节回退/重训流程；未受损 → 重立基线并记录
​

💡 把 PSI 报告并进第 8.2 节监控面板的"漂移层"：它不是独立报表，而是告警体系的一个数据源——>0.25 的判级直接映射到告警分级表。

五、PSI 的两个盲区与补救

1. 均值不变、形状变了：样本从两端向中间收缩（或反之），均值不动，PSI 却会动——这其实是 PSI 的优点；真正的盲区是两维以上的联合漂移：每个单变量 PSI 都稳定，但组合变了（长短输入占比不变、场景×长度联合分布变了）。补救：对高风险维度对加做联合分桶 PSI（桶 = 场景 × 长度档），数量控制在个位数；
2. 慢漂移：每周比基线只挪一点点，单周 PSI 永远 <0.1。补救：滚动基线对比——除"与固定基线比"外，再加"最近 4 周均值 vs 再前 4 周均值"的对比，慢趋势会现形（观点：PSI 回答"变没变"，趋势检验回答"是不是在一直变"）。

本节要点回顾

1. 公式与口径：PSI = Σ(占比差 × ln 占价比)；<0.1 稳定 / 0.1~0.25 观察 / >0.25 显著漂移（风控通行，社区）。
2. 工程细节：等频分桶、桶数 ~10、零桶平滑、基线可辩护且随版本重立。
3. 周度节奏：周一跑矩阵、周四评审、>0.25 当日四步动作（下钻/抽样/复跑/回退或重立）。
4. 盲区：联合漂移加做联合分桶；慢漂移加滚动基线对比。

漂移检测守住"上线之后"，但它只是报警器。报警响了谁来做决定、怎么切流量、多快能回滚——下一章把线上评测与监控的全套防线补齐：影子模式与 A/B、监控面板与告警、回退预案。