2. PSI计算原理：分箱方法与分布对比逻辑

群体稳定性指标（PSI）的核心思想是量化两个概率分布之间的差异。在实际业务中，我们通常将模型评分或特征值离散化（分箱），然后比较每个箱内样本占比的变化。本章将深入解析分箱的三种主流方法，以及实际分布与预期分布的对比逻辑。

2.1 分箱方法：等距、等频与分位数

分箱是将连续变量或离散值划分为若干个区间（箱）的过程。不同的分箱方法会直接影响PSI的计算结果和敏感度。以下是三种最常用的分箱方法对比：

2.2 分箱流程与PSI计算步骤

无论采用哪种分箱方法，PSI的计算都遵循以下标准流程：

1. 确定预期分布（Expected）：通常使用模型训练集或验证集的样本分布作为基准。
2. 确定实际分布（Actual）：使用当前监控周期（如月度、周度）的样本分布。
3. 分箱对齐：使用预期分布的箱边界，对实际分布进行相同的分箱操作。
4. 计算每个箱的占比：
   • 预期占比 \( E_i = \frac{\text{预期样本数}_i}{\text{预期总样本数}} \)
   • 实际占比 \( A_i = \frac{\text{实际样本数}_i}{\text{实际总样本数}} \)
5. 计算每个箱的PSI贡献： \( (A_i - E_i) \times \ln(\frac{A_i}{E_i}) \)
6. 汇总PSI值： \( PSI = \sum_{i=1}^{n} (A_i - E_i) \times \ln(\frac{A_i}{E_i}) \)

2.3 实际分布与预期分布的对比逻辑

PSI的本质是衡量两个分布之间的“距离”。对比逻辑并非简单的数值相减，而是通过加权对数比率来放大差异。下图展示了分箱后分布对比的核心流程：

2.4 对比逻辑的数学本质

从信息论角度看，PSI实际上是相对熵（KL散度）的一种对称化变体。对比逻辑的核心在于：

• 方向性：当 \( A_i > E_i \) 时，该箱的PSI贡献为正，表示实际分布在该区间聚集了更多样本；反之则为负贡献。
• 非线性放大：对数函数 \( \ln(A_i / E_i) \) 对微小差异不敏感，但对较大差异（尤其是当 \( A_i \) 远大于 \( E_i \) 时）会显著放大，从而突出分布漂移。
• 对称性：虽然PSI本身不对称（交换A和E结果不同），但在监控场景中，我们固定E为基准，A为当前，因此方向一致，具有可比性。

2.5 分箱方法选择对PSI的影响

不同的分箱方法会导致PSI值产生显著差异。以下是一个模拟对比示例：

2.6 代码实现示例

以下Python代码展示了使用分位数分箱计算PSI的完整流程：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """
    使用分位数分箱计算PSI
    
    Parameters:
    expected: array-like, 预期分布数据
    actual: array-like, 实际分布数据
    bins: int, 分箱数量（默认10个分位数箱）
    
    Returns:
    psi: float, PSI值
    psi_df: DataFrame, 各箱明细
    """
    # 1. 基于预期分布确定分位数边界
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected, percentiles)
    # 处理边界重复（当数据中有大量重复值时）
    bin_edges = np.unique(bin_edges)
    
    # 2. 对预期和实际分布进行分箱
    expected_bins = np.digitize(expected, bin_edges, right=True)
    actual_bins = np.digitize(actual, bin_edges, right=True)
    
    # 3. 计算每个箱的样本占比
    expected_counts = np.bincount(expected_bins, minlength=len(bin_edges)+1)[1:]
    actual_counts = np.bincount(actual_bins, minlength=len(bin_edges)+1)[1:]
    
    expected_pct = expected_counts / len(expected)
    actual_pct = actual_counts / len(actual)
    
    # 4. 处理零值，避免对数计算错误
    eps = 1e-6
    expected_pct = np.where(expected_pct == 0, eps, expected_pct)
    actual_pct = np.where(actual_pct == 0, eps, actual_pct)
    
    # 5. 计算每个箱的PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    
    # 6. 汇总PSI
    psi = np.sum(psi_components)
    
    # 构建明细DataFrame
    psi_df = pd.DataFrame({
        '箱边界': [f'({bin_edges[i]:.2f}, {bin_edges[i+1]:.2f}]' 
                    for i in range(len(bin_edges)-1)],
        '预期占比': expected_pct,
        '实际占比': actual_pct,
        'PSI贡献': psi_components
    })
    
    return psi, psi_df

# 示例用法
np.random.seed(42)
expected_data = np.random.normal(0, 1, 10000)
actual_data = np.random.normal(0.2, 1.1, 10000)  # 模拟轻微漂移

psi_value, psi_details = calculate_psi(expected_data, actual_data, bins=10)
print(f"PSI值: {psi_value:.4f}")
print(psi_details.head())