2. PSI的数学原理

PSI（Population Stability Index，群体稳定性指数）是衡量模型评分分布随时间变化程度的量化指标。其核心思想是：如果模型在训练集（预期分布）上的评分分布与当前生产环境（实际分布）的评分分布发生显著偏离，则说明模型可能已失效或数据环境已发生漂移。

2.1 PSI的公式推导

PSI 的数学定义基于信息论中的 KL散度（Kullback-Leibler divergence） 的对称化形式。对于两个离散概率分布 P（预期分布）和 Q（实际分布），PSI 的计算公式为：

PSI = Σ ( (Actual_i - Expected_i) * ln(Actual_i / Expected_i) )

其中：

• Expected_i：第 i 个分箱中预期样本占比（通常来自训练集或基准期）
• Actual_i：第 i 个分箱中实际样本占比（来自当前监控期）
• ln：自然对数

公式的推导逻辑：

1. 分箱对齐：将模型评分（如信用评分、概率值）划分为 K 个区间，确保预期分布和实际分布使用相同的分箱边界。
2. 计算占比：分别统计每个分箱中预期样本数和实际样本数，除以各自总样本数得到占比。
3. 逐箱计算：对每个分箱，计算 (实际占比 - 预期占比) × ln(实际占比 / 预期占比)。
4. 求和：将所有分箱的结果累加，得到最终的 PSI 值。

2.2 分箱策略

分箱是 PSI 计算的前置步骤，直接影响指标的敏感性和稳定性。常用的分箱策略有两种：

2.3 PSI值的计算步骤详解

以下通过一个完整示例，展示 PSI 的手动计算过程：

步骤 1：确定分箱边界

假设模型评分范围为 0~100，使用等频分箱策略分为 5 箱。基于训练集（预期分布）的 1000 个样本，确定分箱边界：

步骤 2：统计实际分布

从当前生产环境抽取 800 个样本，使用相同的分箱边界统计：

步骤 3：逐箱计算 PSI 分量

对每个分箱，计算 (Actual_i - Expected_i) × ln(Actual_i / Expected_i)：

步骤 4：求和得到 PSI

PSI = 0.04055 + 0.00000 + 0.01439 + 0.00000 + 0.01439 = 0.06933

2.4 PSI 计算流程可视化

下图展示了从原始评分数据到最终 PSI 值的完整计算流程：

2.5 PSI 计算代码示例

以下 Python 代码展示了 PSI 的完整计算过程，可直接用于模型监控脚本：

import numpy as np

def calculate_psi(expected, actual, bins=10, strategy='equal_freq'):
    """
    计算群体稳定性指数 (PSI)
    
    参数:
        expected: 预期分布（训练集）的评分数组
        actual: 实际分布（当前生产）的评分数组
        bins: 分箱数量，默认10
        strategy: 分箱策略，'equal_freq'（等频）或 'equal_width'（等距）
    
    返回:
        psi_value: PSI 值
        bin_edges: 分箱边界
    """
    # 步骤1：确定分箱边界
    if strategy == 'equal_freq':
        # 等频分箱：基于预期分布的分位数
        bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))
    else:
        # 等距分箱：基于评分范围
        bin_edges = np.linspace(np.min(expected), np.max(expected), bins + 1)
    
    # 步骤2：统计预期分布和实际分布的样本占比
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 转换为占比
    expected_pct = expected_counts / len(expected)
    actual_pct = actual_counts / len(actual)
    
    # 步骤3：平滑处理，避免 ln(0)
    epsilon = 1e-6
    expected_pct = np.clip(expected_pct, epsilon, 1 - epsilon)
    actual_pct = np.clip(actual_pct, epsilon, 1 - epsilon)
    
    # 步骤4：逐箱计算并求和
    psi_value = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return psi_value, bin_edges

# 示例用法
expected_scores = np.random.normal(70, 15, 10000)  # 训练集评分
actual_scores = np.random.normal(65, 18, 8000)     # 生产环境评分

psi, edges = calculate_psi(expected_scores, actual_scores, bins=10, strategy='equal_freq')
print(f"PSI 值: {psi:.4f}")

# 判断漂移程度
if psi < 0.10:
    print("状态：无漂移，模型稳定")
elif psi <= 0.25:
    print("状态：轻度漂移，建议关注")
else:
    print("状态：显著漂移，立即告警！")