4. PSI（群体稳定性指标）计算与解读

PSI（Population Stability Index）是衡量模型评分分布随时间变化程度的量化指标，是批次偏移排查中最核心的监控工具。当模型上线后，若输入数据的分布发生显著变化，模型预测能力将退化，PSI 能第一时间发出预警。

4.1 PSI 公式推导

PSI 基于信息论中的 KL 散度（Kullback-Leibler divergence）构建，衡量两个概率分布之间的差异。其标准计算公式为：

PSI = Σ (P_i - Q_i) * ln(P_i / Q_i)

其中：
- P_i：实际分布（当前批次）在第 i 个分箱中的样本占比
- Q_i：期望分布（训练集/基准集）在第 i 个分箱中的样本占比
- i：分箱索引，i = 1, 2, ..., n

推导逻辑：KL 散度衡量分布 Q 到分布 P 的相对熵，但 KL 散度不对称。PSI 采用对称化处理，将两个方向（P→Q 和 Q→P）的 KL 散度相加后取平均，得到对称性指标：

PSI = KL(P||Q) + KL(Q||P)
     = Σ P_i * ln(P_i/Q_i) + Σ Q_i * ln(Q_i/P_i)
     = Σ (P_i - Q_i) * ln(P_i/Q_i)

实际应用中，PSI 值越小表示分布越稳定，值越大表示偏移越严重。

4.2 分箱策略

分箱是 PSI 计算的前置步骤，分箱方式直接影响 PSI 的敏感度和稳定性。常用分箱策略如下：

推荐实践：

• 分箱数量通常为 10~20 箱，过少会掩盖局部偏移，过多会导致噪声放大。
• 确保每个分箱在基准分布中至少有 5% 的样本，避免分母过小导致 PSI 异常放大。
• 若某分箱在当前批次中样本数为 0，需做平滑处理（如加 0.5 或 1e-6），防止 ln(0) 错误。

4.3 阈值设定（0.1 / 0.2）

行业通用阈值体系如下：

阈值选择依据：

• 0.1：基于统计显著性检验，当 PSI > 0.1 时，两个分布差异在 95% 置信水平下显著。
• 0.2：经验阈值，表示分布偏移已对模型性能产生实质性影响，AUC 或 KS 通常下降 5% 以上。
• 业务调整：高风险场景（如信贷审批、医疗诊断）可收紧至 0.05/0.15；低风险场景（如推荐排序）可放宽至 0.15/0.25。

4.4 PSI 异常时的排查方向

当 PSI 超过阈值时，需系统化排查偏移根因。以下为标准化排查流程：

1. 确认数据质量：检查当前批次是否存在缺失值激增、异常值、数据采集链路故障。
2. 单特征 PSI 分解：计算每个特征的 PSI，定位贡献最大的 Top-K 特征。
3. 特征分布对比：对高贡献特征绘制训练集 vs 当前批次的分布直方图，观察均值/方差/偏度变化。
4. 业务事件关联：排查是否因业务策略调整（如客群下沉、产品改版）、外部环境变化（如政策、季节）导致。
5. 模型性能验证：计算当前批次的 AUC、KS、准确率等指标，确认偏移是否已影响预测效果。
6. 回滚或重训决策：若性能下降超过容忍阈值，启动模型回滚或紧急重训。

4.5 PSI 计算代码示例

以下为 Python 实现 PSI 计算的参考代码，包含分箱、平滑处理和阈值判断：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, bucket_type='quantile'):
    """
    计算群体稳定性指标 PSI

    Parameters:
    - expected: array-like, 基准分布（训练集评分）
    - actual: array-like, 当前分布（上线批次评分）
    - bins: int, 分箱数量
    - bucket_type: str, 'quantile' 等频分箱 或 'uniform' 等距分箱

    Returns:
    - psi: float, PSI 值
    - breakdown: DataFrame, 每个分箱的明细
    """
    # 1. 基于基准分布确定分箱边界
    if bucket_type == 'quantile':
        # 等频分箱：按分位数切分
        cut_points = np.percentile(expected, np.linspace(0, 100, bins + 1))
    else:
        # 等距分箱：按值域均匀切分
        cut_points = np.linspace(np.min(expected), np.max(expected), bins + 1)

    # 2. 确保边界唯一（去重）
    cut_points = np.unique(cut_points)

    # 3. 分箱并计算占比
    expected_binned = np.digitize(expected, cut_points, right=False)
    actual_binned = np.digitize(actual, cut_points, right=False)

    # 统计每个分箱的样本数
    expected_counts = np.bincount(expected_binned, minlength=len(cut_points) + 1)[1:]
    actual_counts = np.bincount(actual_binned, minlength=len(cut_points) + 1)[1:]

    # 4. 计算占比（加平滑防止除零）
    epsilon = 1e-6
    expected_pct = expected_counts / len(expected) + epsilon
    actual_pct = actual_counts / len(actual) + epsilon

    # 5. 计算 PSI
    psi_value = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))

    # 6. 构建明细表
    breakdown = pd.DataFrame({
        'bin_start': cut_points[:-1],
        'bin_end': cut_points[1:],
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'contribution': (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    })

    return psi_value, breakdown

# 使用示例
# psi, detail = calculate_psi(train_scores, online_scores, bins=10)
# print(f"PSI = {psi:.4f}")
# if psi >= 0.2:
#     print("严重偏移，需立即排查")
# elif psi >= 0.1:
#     print("轻度偏移，触发告警")
# else:
#     print("分布稳定")