1. PSI 基础概念：群体稳定性指标

1.1 什么是群体稳定性指标（PSI）

群体稳定性指标（Population Stability Index，PSI） 是衡量两个样本分布之间差异程度的量化指标。在机器学习模型监控中，PSI 用于评估模型输入特征或模型输出分数（预测概率）在时间维度上的分布变化，从而判断模型是否发生“漂移”。

PSI 的核心思想是：如果模型在生产环境中的预测分布与模型训练时的基准分布保持一致，则模型是稳定的；反之，若分布发生显著偏移，则模型可能已失效，需要重新训练或调整。

1.2 PSI 的数学公式与计算逻辑

PSI 的计算基于 信息熵 和 相对熵（KL 散度） 的变体，其数学公式如下：

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

其中：
- P_i = 实际分布（生产环境）在第 i 个分箱中的样本占比
- Q_i = 期望分布（训练基准）在第 i 个分箱中的样本占比
- i   = 分箱序号（1, 2, ..., n）
- ln  = 自然对数

1.3 计算步骤详解

PSI 的计算通常遵循以下 5 个步骤：

1. 分箱（Binning）： 将连续变量或概率值划分为等宽或等频的区间（通常 10 箱）。
2. 计算期望占比（Q_i）： 统计训练集样本在每个分箱中的比例。
3. 计算实际占比（P_i）： 统计生产环境样本在每个分箱中的比例。
4. 计算每个分箱的贡献值： 对每个分箱计算 (P_i - Q_i) * ln(P_i / Q_i)。
5. 求和： 将所有分箱的贡献值相加，得到最终的 PSI 值。

1.4 PSI 计算示例表

以下是一个 5 箱的 PSI 计算示例，帮助理解每个分箱的贡献：

1.5 PSI 计算逻辑流程图

1.6 PSI 的行业解读标准

在实际工程中，PSI 值通常按照以下阈值区间进行解读：

1.7 PSI 与 KL 散度的关系

PSI 与 KL 散度（Kullback-Leibler Divergence）密切相关，但存在关键区别：

• KL 散度： D_KL(P || Q) = Σ P_i * ln(P_i / Q_i)，衡量用 Q 分布近似 P 分布时的信息损失。
• PSI： 是对称化的 KL 散度变体，同时考虑了 P 相对于 Q 和 Q 相对于 P 的差异。

数学上，PSI 可以看作两个 KL 散度的和：

PSI = D_KL(P || Q) + D_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 比 KL 散度更适合作为监控指标，因为它不依赖于哪个分布被当作“基准”。

1.8 PSI 的适用场景

1.9 PSI 计算代码示例（Python）

import numpy as np

def calculate_psi(expected, actual, bins=10):
    """
    计算群体稳定性指标（PSI）
    
    Parameters:
    expected: array-like, 训练集（基准）的预测分数
    actual: array-like, 生产环境的预测分数
    bins: int, 分箱数量
    
    Returns:
    psi: float, PSI 值
    """
    # 1. 确定分箱边界（基于期望分布）
    bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))
    
    # 2. 计算期望占比
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    expected_pct = expected_counts / len(expected)
    
    # 3. 计算实际占比
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    actual_pct = actual_counts / len(actual)
    
    # 4. 处理零值（加极小值避免除零）
    expected_pct = np.clip(expected_pct, 0.0001, 1.0)
    actual_pct = np.clip(actual_pct, 0.0001, 1.0)
    
    # 5. 计算 PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return psi

# 示例用法
expected_scores = np.random.beta(2, 5, 10000)  # 训练集分布
actual_scores = np.random.beta(2.5, 4.5, 10000)  # 生产环境分布

psi_value = calculate_psi(expected_scores, actual_scores, bins=10)
print(f"PSI 值: {psi_value:.4f}")