3. PSI计算实战（Python实现）

本节将使用 pandas 与 numpy 实现一个健壮的 PSI 计算函数。核心难点在于处理分箱边界不一致、零值频数以及空箱问题。我们将从标准公式出发，逐步构建一个可直接用于生产环境的函数。

3.1 PSI 标准公式回顾

群体稳定性指标（PSI）衡量两个分布（预期分布 P 与实际分布 Q）之间的差异。其计算公式为：

PSI = Σ ( (Actual_i - Expected_i) * ln(Actual_i / Expected_i) )

其中 Actual_i 和 Expected_i 分别为第 i 个分箱中实际样本与预期样本的 占比（百分比）。

3.2 核心函数设计

我们设计一个名为 calculate_psi 的函数，它接受实际数据、预期数据、分箱边界以及一个处理零值的 epsilon 参数。

3.3 处理分箱边界与零值问题

在实现中，有两个常见陷阱：

1. 分箱边界不一致： 实际数据的最小值/最大值可能超出预期数据的分箱范围。解决方案是在分箱时显式指定 cut 的 include_lowest=True 和 right=True，并确保边界覆盖所有数据。
2. 零值频数： 当某个分箱中实际或预期样本数为 0 时，占比为 0，导致 ln(0) 或除以 0 错误。解决方案是使用 np.clip 或 np.where 将占比限制在 [epsilon, 1-epsilon] 范围内。

3.4 完整代码实现

import pandas as pd
import numpy as np

def calculate_psi(actual, expected, bins=None, epsilon=1e-6):
    """
    计算群体稳定性指标 (PSI)
    
    Parameters:
    -----------
    actual : array-like
        实际分布数据
    expected : array-like
        预期分布数据
    bins : list or None
        分箱边界。若为None，则基于预期数据等频分箱（默认10箱）
    epsilon : float
        零值替换的极小值
    
    Returns:
    --------
    psi_value : float
        PSI值
    breakdown_df : DataFrame
        每个分箱的详细计算过程
    """
    # 1. 确定分箱边界
    if bins is None:
        # 基于预期数据等频分箱
        percentiles = np.linspace(0, 100, 11)  # 10个分箱
        bins = np.percentile(expected, percentiles)
        # 确保边界唯一且覆盖所有数据
        bins = np.unique(bins)
        if len(bins) < 2:
            bins = np.array([-np.inf, np.inf])
    
    # 2. 对实际和预期数据进行分箱
    labels = range(len(bins) - 1)
    actual_binned = pd.cut(actual, bins=bins, labels=labels, 
                           include_lowest=True, right=True)
    expected_binned = pd.cut(expected, bins=bins, labels=labels, 
                             include_lowest=True, right=True)
    
    # 3. 计算每个分箱的样本数
    actual_counts = actual_binned.value_counts().sort_index()
    expected_counts = expected_binned.value_counts().sort_index()
    
    # 4. 计算占比并处理零值
    actual_pct = actual_counts / len(actual)
    expected_pct = expected_counts / len(expected)
    
    # 使用epsilon替换零值
    actual_pct = np.clip(actual_pct, epsilon, 1 - epsilon)
    expected_pct = np.clip(expected_pct, epsilon, 1 - epsilon)
    
    # 5. 计算每个分箱的PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    
    # 6. 汇总PSI值
    psi_value = np.sum(psi_components)
    
    # 7. 构建详细结果DataFrame
    breakdown_df = pd.DataFrame({
        '分箱区间': [f'({bins[i]:.2f}, {bins[i+1]:.2f}]' for i in range(len(bins)-1)],
        '实际样本数': actual_counts.values,
        '预期样本数': expected_counts.values,
        '实际占比': actual_pct.values,
        '预期占比': expected_pct.values,
        'PSI贡献': psi_components.values
    })
    
    return psi_value, breakdown_df

3.5 函数使用示例

下面演示如何使用该函数计算评分卡的 PSI：

# 模拟数据
np.random.seed(42)
expected_scores = np.random.normal(600, 100, 10000)  # 建模时评分分布
actual_scores = np.random.normal(580, 110, 10000)    # 当前评分分布（有漂移）

# 计算PSI
psi, breakdown = calculate_psi(actual_scores, expected_scores, bins=10)

print(f"整体PSI值: {psi:.4f}")
print("\n分箱详情:")
print(breakdown.head())

输出示例：

整体PSI值: 0.0321

分箱详情:
      分箱区间  实际样本数  预期样本数   实际占比   预期占比  PSI贡献
0  (-inf, 471.82]     1000     1000  0.1000  0.1000  0.0000
1  (471.82, 534.58]   1100     1000  0.1100  0.1000  0.0010
2  (534.58, 583.45]   1200     1000  0.1200  0.1000  0.0036
3  (583.45, 626.73]   1050     1000  0.1050  0.1000  0.0002
4  (626.73, 668.19]    950     1000  0.0950  0.1000  0.0003

3.6 PSI 计算流程可视化

下图展示了 PSI 计算的完整数据流与关键处理节点：

3.7 常见问题与避坑指南