3. PSI计算实战：使用Python计算PSI

群体稳定性指标（PSI）是信用评分模型监控中最核心的量化指标之一。本章通过完整的Python代码实战，从数据分箱到最终PSI值计算，逐步拆解每个环节的数学逻辑与实现细节，并深入解读计算结果的实际业务含义。

3.1 PSI计算全流程概览

PSI的计算本质是比较两个分布（预期分布 vs 实际分布）在每个分箱上的偏移程度。标准计算流程包含以下五个步骤：

1. 数据准备：获取模型评分（或特征值）的预期样本（开发样本）与实际样本（监控样本）。
2. 分箱（Binning）：对评分区间进行等距或等频分箱，通常建议10~20箱。
3. 计算占比：分别计算每个分箱在预期样本和实际样本中的占比。
4. 计算WOE：计算每个分箱的权重证据（WOE = ln(实际占比 / 预期占比)）。
5. 计算PSI：对每个分箱的 (实际占比 - 预期占比) × WOE 求和，得到最终PSI值。

3.2 Python代码实现

以下代码实现了完整的PSI计算函数，包含等距分箱、占比计算、WOE计算和PSI聚合。代码采用纯NumPy/Pandas实现，不依赖第三方评分库，便于理解底层逻辑。

import pandas as pd
import numpy as np

def calculate_psi(expected, actual, bins=10, bin_type='equal_width'):
    """
    计算群体稳定性指标（PSI）
    
    参数:
        expected: array-like, 预期样本（开发样本）的评分
        actual: array-like, 实际样本（监控样本）的评分
        bins: int, 分箱数量，默认10
        bin_type: str, 'equal_width' 等距分箱 或 'equal_freq' 等频分箱
    
    返回:
        psi_value: float, PSI值
        psi_table: DataFrame, 每个分箱的详细计算结果
    """
    # 合并数据确定分箱边界
    all_scores = np.concatenate([expected, actual])
    
    if bin_type == 'equal_width':
        # 等距分箱：根据最小最大值均匀切分
        bin_edges = np.linspace(np.min(all_scores), np.max(all_scores), bins + 1)
    elif bin_type == 'equal_freq':
        # 等频分箱：根据预期样本的分位数切分
        bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))
        # 确保边界唯一
        bin_edges = np.unique(bin_edges)
    else:
        raise ValueError("bin_type 必须为 'equal_width' 或 'equal_freq'")
    
    # 分箱并计算频数
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 计算占比（加微小平滑避免除零）
    epsilon = 1e-10
    expected_pct = expected_counts / (len(expected) + epsilon)
    actual_pct = actual_counts / (len(actual) + epsilon)
    
    # 计算WOE和PSI分量
    woe = np.log((actual_pct + epsilon) / (expected_pct + epsilon))
    psi_components = (actual_pct - expected_pct) * woe
    psi_value = np.sum(psi_components)
    
    # 构建结果表格
    psi_table = pd.DataFrame({
        'bin_start': bin_edges[:-1],
        'bin_end': bin_edges[1:],
        'expected_count': expected_counts,
        'actual_count': actual_counts,
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'woe': woe,
        'psi_component': psi_components
    })
    
    return psi_value, psi_table

# 示例数据生成
np.random.seed(42)
expected_scores = np.random.normal(600, 100, 10000)  # 预期样本：均值600，标准差100
actual_scores = np.random.normal(620, 110, 10000)    # 实际样本：均值620，标准差110

# 计算PSI
psi_value, psi_table = calculate_psi(expected_scores, actual_scores, bins=10)
print(f"整体PSI值: {psi_value:.4f}")
print("\n各分箱详细结果（前5行）:")
print(psi_table.head())

3.3 计算结果解读

运行上述代码后，输出结果如下表所示。整体PSI值为 0.0327，远低于0.1的阈值，表明当前模型评分分布与开发样本相比无显著偏移。

3.4 分箱策略对比

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

3.5 常见问题与避坑指南

• 空箱处理： 当某个分箱在实际样本中无数据时，占比为0，导致WOE为负无穷。解决方案是加入极小平滑值（如1e-10）或合并相邻空箱。
• 分箱数量选择： 10箱是行业标准，但若样本量较小（<5000），建议减少至5~8箱；若样本量极大（>100万），可增加至15~20箱。
• 边界稳定性： 生产环境中建议固定分箱边界（使用开发样本的边界），避免每次监控时边界漂移导致PSI失真。
• WOE符号解读： WOE为正表示该箱人群占比上升（风险可能恶化），WOE为负表示占比下降（风险可能改善），但需结合业务逻辑判断。

3.6 生产环境代码封装建议

在实际生产系统中，PSI计算通常需要封装为可重复调用的监控模块。以下给出一个更健壮的生产级封装示例：

class PSIMonitor:
    """生产级PSI监控器"""
    
    def __init__(self, bin_edges=None, bins=10, smoothing=1e-10):
        self.bin_edges = bin_edges  # 固定分箱边界
        self.bins = bins
        self.smoothing = smoothing
        self.history = []  # 存储历史PSI记录
    
    def fit(self, expected):
        """基于预期样本拟合分箱边界"""
        self.bin_edges = np.percentile(expected, np.linspace(0, 100, self.bins + 1))
        self.bin_edges = np.unique(self.bin_edges)
        self.expected_pct = self._get_distribution(expected)
        return self
    
    def _get_distribution(self, data):
        """计算数据在固定边界上的分布"""
        counts, _ = np.histogram(data, bins=self.bin_edges)
        return counts / (len(data) + self.smoothing)
    
    def calculate(self, actual, timestamp=None):
        """计算当前周期PSI"""
        actual_pct = self._get_distribution(actual)
        woe = np.log((actual_pct + self.smoothing) / (self.expected_pct + self.smoothing))
        psi = np.sum((actual_pct - self.expected_pct) * woe)
        
        record = {'timestamp': timestamp, 'psi': psi}
        self.history.append(record)
        return psi

# 使用示例
monitor = PSIMonitor(bins=10)
monitor.fit(expected_scores)

# 模拟多个监控周期
for month in range(1, 13):
    new_actual = np.random.normal(600 + month * 2, 100, 10000)
    psi = monitor.calculate(new_actual, timestamp=f"2024-{month:02d}")
    print(f"{month:02d}月 PSI: {psi:.4f}")