3. PSI 计算实战：使用 Python 计算 PSI 的完整代码实现

本节将带领你从零开始，使用 Python 实现 PSI 的完整计算流程。我们将深入探讨两种主流分箱策略——等距分箱与等频分箱，并通过实际代码对比它们对 PSI 结果的直接影响。

3.1 PSI 计算核心步骤

PSI 的计算本质上是对比两个分布（预期分布与实际分布）在各分箱上的比例变化。其标准计算流程如下：

1. 数据分箱：将连续变量或离散变量划分为 K 个区间（箱）。
2. 计算预期占比：统计每个箱中样本数占总样本数的比例（基准分布）。
3. 计算实际占比：统计新数据在每个箱中的样本占比（当前分布）。
4. 计算各箱 PSI 分量：对每个箱，计算 (实际占比 - 预期占比) * ln(实际占比 / 预期占比)。
5. 汇总 PSI：将所有箱的 PSI 分量求和，得到最终的 PSI 值。

3.2 完整 Python 代码实现

以下代码实现了 PSI 计算的核心函数，并支持自定义分箱策略：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, strategy='equal_width'):
    """
    计算 PSI（群体稳定性指标）
    
    参数:
        expected: array-like, 预期分布（基准数据）
        actual: array-like, 实际分布（当前数据）
        bins: int, 分箱数量
        strategy: str, 分箱策略 ('equal_width' 或 'equal_freq')
    
    返回:
        psi: float, PSI 值
        bin_df: DataFrame, 各箱明细
    """
    # 合并数据确定分箱边界
    all_data = np.concatenate([expected, actual])
    
    if strategy == 'equal_width':
        # 等距分箱：在数据范围内均匀划分
        bin_edges = np.linspace(np.min(all_data), np.max(all_data), bins + 1)
    elif strategy == 'equal_freq':
        # 等频分箱：按分位数划分，使每箱样本数大致相等
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(all_data, percentiles)
        # 确保边界唯一
        bin_edges = np.unique(bin_edges)
    else:
        raise ValueError("strategy 必须是 'equal_width' 或 'equal_freq'")
    
    # 计算各箱样本数
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 计算占比（添加极小值避免除零）
    epsilon = 1e-6
    expected_pct = expected_counts / len(expected) + epsilon
    actual_pct = actual_counts / len(actual) + epsilon
    
    # 计算各箱 PSI 分量
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi = np.sum(psi_components)
    
    # 构建明细 DataFrame
    bin_df = pd.DataFrame({
        'bin_start': bin_edges[:-1],
        'bin_end': bin_edges[1:],
        'expected_count': expected_counts,
        'actual_count': actual_counts,
        'expected_pct': expected_pct - epsilon,
        'actual_pct': actual_pct - epsilon,
        'psi_component': psi_components
    })
    
    return psi, bin_df

# 示例用法
np.random.seed(42)
expected_data = np.random.normal(0, 1, 10000)
actual_data = np.random.normal(0.2, 1.2, 10000)

psi_value, details = calculate_psi(expected_data, actual_data, bins=10, strategy='equal_width')
print(f"PSI (等距分箱): {psi_value:.4f}")

psi_value_eq, details_eq = calculate_psi(expected_data, actual_data, bins=10, strategy='equal_freq')
print(f"PSI (等频分箱): {psi_value_eq:.4f}")

3.3 分箱策略对比：等距 vs 等频

分箱策略直接影响 PSI 的敏感性和稳定性。下表总结了两种策略的核心差异：

3.4 分箱策略对 PSI 的影响可视化

下图直观展示了两种分箱策略下，预期分布与实际分布的对比情况：

3.5 实战对比：不同分箱策略下的 PSI 差异

我们使用模拟数据来观察两种分箱策略的实际效果。假设基准数据服从标准正态分布，当前数据均值偏移 0.2、标准差扩大 1.2 倍：

# 模拟数据
np.random.seed(42)
expected = np.random.normal(0, 1, 50000)
actual = np.random.normal(0.2, 1.2, 50000)

# 等距分箱
psi_width, df_width = calculate_psi(expected, actual, bins=10, strategy='equal_width')
print(f"等距分箱 PSI: {psi_width:.4f}")
print(df_width[['bin_start', 'bin_end', 'expected_pct', 'actual_pct', 'psi_component']].to_string())

print("\n" + "="*60 + "\n")

# 等频分箱
psi_freq, df_freq = calculate_psi(expected, actual, bins=10, strategy='equal_freq')
print(f"等频分箱 PSI: {psi_freq:.4f}")
print(df_freq[['bin_start', 'bin_end', 'expected_pct', 'actual_pct', 'psi_component']].to_string())

运行上述代码，你会观察到以下典型结果：

• 等距分箱：PSI 值通常偏高（例如 0.15~0.25），因为边缘箱样本稀疏，占比微小变化被对数放大。
• 等频分箱：PSI 值更稳定（例如 0.08~0.12），各箱样本均匀，对分布偏移的度量更平滑。

3.6 分箱策略选择决策流程

下图总结了在实际项目中如何选择分箱策略：

3.7 最佳实践与避坑指南

• 分箱数量建议：通常 5~20 箱。箱数太少会丢失分布细节，箱数太多则每个箱样本过少，PSI 不稳定。
• 零占比处理：始终添加极小值（如 1e-6）或使用平滑技术，避免 ln(0) 导致计算失败。
• 业务对齐：如果模型输出有业务含义的阈值（如信用评分 600 分是拒绝线），优先使用等距分箱并确保边界与业务阈值对齐。
• 交叉验证：在正式监控上线前，使用历史数据模拟 PSI 计算，对比两种分箱策略的结果差异，选择最稳定的方案。
• 动态调整：监控过程中如果发现某箱长期样本过少，应动态合并相邻箱，保持计算稳定性。