3. PSI 指标原理：群体稳定性指标

群体稳定性指标（Population Stability Index, PSI）是衡量两个样本分布之间差异程度的统计量，在风控、营销和模型监控领域被广泛用于检测特征或模型分数的分布漂移。PSI 的核心思想是：如果模型在训练集上的预测分布与当前生产环境中的预测分布一致，则模型表现稳定；反之，分布差异越大，模型失效的风险越高。

3.1 数学定义与公式

PSI 的计算基于信息论中的 KL 散度（Kullback-Leibler Divergence）的对称化变体。其标准公式如下：

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

其中：

• P_i：实际分布（Actual）在第 i 个分箱中的样本占比，通常指当前生产数据或验证数据。
• Q_i：期望分布（Expected）在第 i 个分箱中的样本占比，通常指训练数据或基准数据。
• ln(P_i / Q_i)：两个占比比值的自然对数，衡量该分箱内的偏移程度。
• Σ：对所有分箱求和，得到最终的 PSI 值。

3.2 分箱策略

PSI 的计算依赖分箱操作，不同的分箱策略会直接影响 PSI 的敏感度和稳定性。常用的分箱方法有三种：

分箱数量 N 的选择也至关重要。N 过小会掩盖局部漂移，N 过大则容易引入噪声。经验上，对于模型分数（如 0~1 的概率值），推荐使用 10~20 个分箱；对于连续特征，可根据样本量调整，通常取 10~30 个分箱。

3.3 PSI 阈值解读

PSI 的数值没有绝对的上限，但在实际工程中，行业通用阈值如下：

3.4 PSI 计算流程与可视化

下图展示了 PSI 从数据输入到告警决策的完整计算流程：

3.5 代码实现示例

以下 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':
        # 等频分箱：使用分位数
        percentiles = np.linspace(0, 100, bins + 1)
        breakpoints = np.percentile(expected, percentiles)
    else:
        # 等距分箱：均匀划分
        breakpoints = np.linspace(
            np.min(expected), np.max(expected), bins + 1
        )
    
    # 2. 确保边界唯一，避免除零
    breakpoints = np.unique(breakpoints)
    
    # 3. 对两个分布分别分箱
    expected_bins = np.digitize(expected, breakpoints, right=True)
    actual_bins = np.digitize(actual, breakpoints, right=True)
    
    # 4. 计算每个分箱的占比
    expected_counts = np.bincount(expected_bins, minlength=len(breakpoints)+1)[1:]
    actual_counts = np.bincount(actual_bins, minlength=len(breakpoints)+1)[1:]
    
    # 添加平滑因子，防止 ln(0)
    epsilon = 1e-6
    expected_pct = expected_counts / len(expected) + epsilon
    actual_pct = actual_counts / len(actual) + epsilon
    
    # 5. 计算 PSI
    psi_values = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi = np.sum(psi_values)
    
    # 6. 构建详细分解表
    breakdown = pd.DataFrame({
        '分箱下限': breakpoints[:-1],
        '分箱上限': breakpoints[1:],
        '基准占比': expected_pct,
        '实际占比': actual_pct,
        '单项贡献': psi_values
    })
    
    return psi, breakdown

# 示例用法
np.random.seed(42)
train_scores = np.random.beta(2, 5, 10000)  # 基准分布
prod_scores = np.random.beta(2.5, 4.5, 10000)  # 生产分布（轻度漂移）

psi_value, detail = calculate_psi(train_scores, prod_scores, bins=10)
print(f"PSI = {psi_value:.4f}")

if psi_value < 0.1:
    print("结论：无显著漂移，模型稳定")
elif psi_value < 0.25:
    print("结论：轻度漂移，建议关注")
else:
    print("结论：严重漂移，立即告警！")