3. PSI计算步骤：数据分箱方法（等距、等频、分位数）、计算各箱占比、计算PSI值、汇总结果

PSI（Population Stability Index，群体稳定性指标）的计算核心在于比较两个分布（通常为基准分布与当前分布）在各区间上的占比差异。整个计算流程可归纳为四个关键步骤：数据分箱、计算各箱占比、计算PSI值、汇总结果。其中，数据分箱方法的选择直接影响PSI的敏感度和稳定性。

3.1 数据分箱方法

分箱是将连续变量或离散变量划分为若干个区间（箱）的过程。在PSI计算中，分箱的目的是将复杂的分布简化为可比较的区间占比。常用的分箱方法有三种：等距分箱、等频分箱和分位数分箱。

3.2 计算各箱占比

确定分箱边界后，分别计算基准分布和当前分布中落在每个箱内的样本占比。占比计算公式为：

箱占比 = 该箱内样本数 / 总样本数

假设我们将某个评分变量分为5个箱，基准分布和当前分布的占比计算示例如下：

3.3 计算PSI值

PSI的计算公式基于每个箱的占比差异，通过以下公式逐箱计算并累加：

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

其中：

• P_i：基准分布在第i个箱的占比
• Q_i：当前分布在第i个箱的占比
• ln：自然对数

继续以上述表格数据为例，逐箱计算PSI贡献值：

最终PSI值 = 0.0067 + 0.0004 + 0.0003 + 0.0021 + 0.0013 = 0.0108。

3.4 汇总结果

汇总结果阶段需要将计算出的PSI值与预设的告警阈值进行比较，并输出监控结论。常见的PSI阈值判断标准如下：

3.5 PSI计算全流程SVG图

下图展示了从原始数据输入到最终告警决策的完整PSI计算链路：

3.6 分箱方法对比SVG图

为了更直观地理解三种分箱方法的差异，下图展示了同一组数据在不同分箱策略下的区间划分效果：

3.7 代码实现示例

以下Python代码展示了使用等频分箱计算PSI的完整实现：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, method='quantile'):
    """
    计算PSI值
    
    Parameters:
    expected: array-like, 基准分布数据
    actual: array-like, 当前分布数据
    bins: int, 分箱数量
    method: str, 分箱方法 ('quantile', 'equal_width', 'equal_freq')
    
    Returns:
    psi: float, PSI值
    bin_details: DataFrame, 各箱详细占比信息
    """
    # 1. 确定分箱边界
    if method == 'quantile':
        # 分位数分箱
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(expected, percentiles)
    elif method == 'equal_width':
        # 等距分箱
        bin_edges = np.linspace(np.min(expected), np.max(expected), bins + 1)
    else:
        # 等频分箱（默认）
        bin_edges = np.percentile(expected, np.linspace(0, 100, 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. 平滑处理（避免除零）
    epsilon = 1e-6
    expected_pct = np.clip(expected_pct, epsilon, 1 - epsilon)
    actual_pct = np.clip(actual_pct, epsilon, 1 - epsilon)
    
    # 4. 计算PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    # 5. 汇总结果
    bin_details = pd.DataFrame({
        'bin_start': bin_edges[:-1],
        'bin_end': bin_edges[1:],
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'psi_contribution': (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    })
    
    return psi, bin_details

# 使用示例
expected_data = np.random.normal(500, 100, 10000)
actual_data = np.random.normal(520, 110, 10000)

psi_value, details = calculate_psi(expected_data, actual_data, bins=10, method='quantile')
print(f"PSI值: {psi_value:.4f}")
print(f"稳定性等级: {'稳定' if psi_value < 0.10 else '轻度漂移' if psi_value < 0.25 else '严重漂移'}")
print("\n各箱详情:")
print(details.head())