3. PSI 计算步骤详解：分箱、计算占比、计算对数比、求和

群体稳定性指标（PSI）的计算过程严格遵循四个步骤：分箱 → 计算占比 → 计算对数比 → 加权求和。每一步都直接影响最终 PSI 值的准确性与可解释性。下面我们以“模型预测分数”为例，逐步拆解每个环节的数学操作与工程实现。

3.1 步骤一：分箱（Binning）

分箱是将连续型预测分数或概率值划分为若干个离散区间（箱）。分箱方式决定了 PSI 对分布偏移的敏感度。

分箱完成后，我们得到 K 个箱（通常 K=10 或 K=20）。每个箱对应一个区间 [b_{i-1}, b_i)，其中 i = 1, 2, …, K。

3.2 步骤二：计算占比（Proportion）

分别计算期望分布（训练集）和实际分布（当前监控集）在每个箱中的样本占比。

设训练集总样本数为 N_train，当前监控集总样本数为 N_current。对于第 i 个箱：

• 期望占比：P_i = count_train,i / N_train
• 实际占比：Q_i = count_current,i / N_current

3.3 步骤三：计算对数比（Log Ratio）

对每个箱，计算实际占比与期望占比的自然对数比：

LogRatio_i = ln(Q_i / P_i)

这个对数比衡量了每个箱中分布偏移的方向与幅度：

• 当 Q_i > P_i 时，对数比 > 0，表示该箱中当前样本占比增加（偏移正向）
• 当 Q_i < P_i 时，对数比 < 0，表示该箱中当前样本占比减少（偏移负向）
• 当 Q_i = P_i 时，对数比 = 0，表示该箱分布无变化

3.4 步骤四：加权求和（Weighted Sum）

将每个箱的占比差与对数比相乘，然后对所有箱求和，得到最终的 PSI 值：

PSI = Σ_{i=1}^{K} (Q_i - P_i) * ln(Q_i / P_i)

这个公式可以理解为：每个箱的贡献 = 占比变化幅度 × 变化方向的对数度量。所有箱的贡献累加后，得到一个非负的标量值。

3.5 完整计算示例（Python 代码）

下面给出一个完整的 PSI 计算函数，包含平滑处理与分箱逻辑：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, epsilon=1e-6):
    """
    计算群体稳定性指标 (PSI)
    
    参数:
        expected: array-like, 训练集（期望分布）的预测分数
        actual: array-like, 当前监控集（实际分布）的预测分数
        bins: int 或 list, 分箱数量或自定义分箱边界
        epsilon: float, 平滑常数，防止 log(0)
    
    返回:
        psi: float, PSI 值
        bin_edges: array, 分箱边界
        contributions: array, 每个箱的 PSI 贡献
    """
    # 1. 分箱：基于期望分布的分位数确定边界
    if isinstance(bins, int):
        bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))
    else:
        bin_edges = np.array(bins)
    
    # 2. 计算占比（加平滑）
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    expected_pct = (expected_counts + epsilon) / (expected.sum() + epsilon * len(expected_counts))
    actual_pct = (actual_counts + epsilon) / (actual.sum() + epsilon * len(actual_counts))
    
    # 3. 计算对数比
    log_ratio = np.log(actual_pct / expected_pct)
    
    # 4. 加权求和
    contributions = (actual_pct - expected_pct) * log_ratio
    psi = np.sum(contributions)
    
    return psi, bin_edges, contributions

# 使用示例
train_scores = np.random.normal(0.5, 0.15, 10000)
current_scores = np.random.normal(0.55, 0.18, 10000)

psi_value, edges, contribs = calculate_psi(train_scores, current_scores, bins=10)
print(f"PSI = {psi_value:.4f}")

3.6 各步骤常见问题与排查