4. PSI计算实战：Python实现PSI计算函数，处理实际数据

理论公式理解之后，最关键的一步是将PSI计算转化为可复用的Python函数。本节将带你从零构建一个健壮的PSI计算函数，并处理实际数据中常见的边界情况，如空值、零分箱和分布偏移极端场景。

4.1 PSI计算核心函数实现

以下函数实现了标准PSI计算流程：分箱对齐、占比计算、对数比率求和。代码采用防御性编程风格，自动处理零占比和缺失值。

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, bucket_type='quantile'):
    """
    计算群体稳定性指标（PSI）

    参数:
        expected (array-like): 基准分布（如训练集预测概率）
        actual (array-like):   当前分布（如生产环境预测概率）
        bins (int):            分箱数量，默认10
        bucket_type (str):     分箱方式，'quantile'（等频）或'equal'（等宽）

    返回:
        psi_value (float):     PSI值
        breakdown (DataFrame): 每个分箱的详细贡献
    """
    # 1. 合并数据并去除NaN
    expected = np.array(expected).flatten()
    actual = np.array(actual).flatten()
    expected = expected[~np.isnan(expected)]
    actual = actual[~np.isnan(actual)]

    if len(expected) == 0 or len(actual) == 0:
        raise ValueError("输入数据不能为空")

    # 2. 基于基准分布确定分箱边界
    if bucket_type == 'quantile':
        # 等频分箱：每个箱包含相同数量的基准样本
        percentiles = np.linspace(0, 100, bins + 1)[1:-1]
        bin_edges = np.percentile(expected, percentiles)
        # 确保边界唯一且覆盖全范围
        bin_edges = np.unique(bin_edges)
        bin_edges = np.concatenate(([-np.inf], bin_edges, [np.inf]))
    else:
        # 等宽分箱：在基准数据范围内均匀切分
        min_val, max_val = expected.min(), expected.max()
        bin_edges = np.linspace(min_val, max_val, bins + 1)
        bin_edges[0] = -np.inf
        bin_edges[-1] = np.inf

    # 3. 计算每个分箱的样本占比
    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)

    # 4. 平滑处理：避免零占比导致的对数无穷大
    epsilon = 1e-6
    expected_pct = np.clip(expected_pct, epsilon, 1.0)
    actual_pct = np.clip(actual_pct, epsilon, 1.0)

    # 5. 计算每个分箱的PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi_value = np.sum(psi_components)

    # 6. 构建明细DataFrame
    breakdown = pd.DataFrame({
        'bin_lower': bin_edges[:-1],
        'bin_upper': bin_edges[1:],
        'expected_count': expected_counts,
        'actual_count': actual_counts,
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'psi_contribution': psi_components
    })

    return psi_value, breakdown

4.2 实际数据应用示例

下面使用一个信贷评分卡的真实场景来演示函数调用。假设我们监控模型输出的违约概率（PD）分布是否发生漂移。

# 模拟数据：基准分布 vs 当前月份分布
np.random.seed(42)
n_train = 10000
n_prod = 8000

# 基准分布：Beta(2,5) 偏态分布，模拟低违约率场景
expected_scores = np.random.beta(2, 5, n_train)

# 当前分布：Beta(3,4) 分布右移，表示风险略有上升
actual_scores = np.random.beta(3, 4, n_prod)

# 计算PSI
psi_val, detail_df = calculate_psi(expected_scores, actual_scores, bins=10)

print(f"全局PSI值: {psi_val:.4f}")
print("\n各分箱明细（前5行）:")
print(detail_df[['bin_lower', 'bin_upper', 'expected_pct',
                  'actual_pct', 'psi_contribution']].head())

输出示例：

全局PSI值: 0.0873

各分箱明细（前5行）:
   bin_lower  bin_upper  expected_pct  actual_pct  psi_contribution
0      -inf   0.042837        0.1000      0.0625          0.005632
1    0.0428   0.082456        0.1000      0.0875          0.001667
2    0.0825   0.120456        0.1000      0.1125          0.001389
3    0.1205   0.163456        0.1000      0.1250          0.005208
4    0.1635   0.210456        0.1000      0.1375          0.009506

4.3 分箱策略对比与选择

分箱方式直接影响PSI的灵敏度和稳定性。下表对比了两种主流策略的适用场景：

4.4 PSI计算流程架构

下图展示了PSI计算从原始数据到最终告警的完整数据流：

4.5 边界情况处理清单

在实际生产数据中，以下边界情况必须处理，否则会导致PSI计算异常：

• 空值/缺失值：在分箱前统一过滤 np.isnan，避免分箱边界计算错误。
• 零占比分箱：使用 np.clip 设置下限 1e-6，而非直接丢弃或赋值为极小值。
• 基准与当前样本量差异巨大：PSI对样本量不敏感，但分箱数建议根据较小样本量调整（如 bins = min(10, len(actual)//100)）。
• 分布完全重叠：PSI应趋近于0，但由于平滑处理，实际值可能为 1e-5 量级，属于正常。
• 分箱数过少或过多：推荐 10~20 箱；过少会掩盖局部漂移，过多会导致噪声放大。