📚 课程目录第4章 PSI计算实战

4. PSI计算实战：使用Python计算PSI值、分箱策略与代码实现

4.1 群体稳定性指标（PSI）核心原理

PSI（Population Stability Index）是风控模型监控中最常用的量化指标，用于衡量两个样本分布之间的差异程度。其核心思想是通过比较预期分布（通常为建模时的训练集或基准期）与实际分布（当前生产数据或监控期）在各分箱上的占比变化，来评估模型输入特征或评分结果的稳定性。

PSI计算公式：

PSI = Σ (实际占比_i - 预期占比_i) * ln(实际占比_i / 预期占比_i)

其中，i 表示第 i 个分箱，实际占比_i 为当前监控期数据在该箱中的样本比例，预期占比_i 为基准期数据在该箱中的样本比例。

PSI阈值参考：

PSI < 0.1：分布无显著变化，模型稳定。
0.1 ≤ PSI < 0.25：分布轻度偏移，需关注。
PSI ≥ 0.25：分布显著偏移，需立即排查并考虑模型重建。

4.2 分箱策略详解

PSI计算对分箱方式高度敏感。不合理的分箱可能导致PSI值失真，掩盖或夸大分布漂移。以下是三种主流分箱策略的对比：

策略	原理	优点	缺点	适用场景
等频分箱	每箱包含相同数量的样本	避免空箱；对密集区域敏感	边界值可能不直观	评分卡分数、概率输出
等距分箱	每箱具有相同的数值区间宽度	边界清晰、易于解释	可能产生空箱或样本稀疏箱	年龄、收入等连续变量
自定义分箱	基于业务知识或统计检验设定断点	灵活、可融入业务逻辑	主观性强、维护成本高	有明确业务阈值的特征

⚠️ 分箱避坑指南：

箱数选择：通常建议 5~20 箱。箱数过少会丢失分布细节，箱数过多会导致噪声放大。
空箱处理：若某箱在基准期或监控期占比为0，会导致 ln(0) 计算异常。建议对占比为0的箱赋予极小值（如 0.0001）或合并相邻箱。
一致性原则：基准期和监控期必须使用完全相同的分箱边界，否则PSI计算无效。

4.3 Python代码实现：PSI计算函数

以下代码实现了完整的PSI计算函数，支持等频分箱和等距分箱两种策略，并包含空箱保护机制。

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, bin_type='equal_frequency'):
    """
    计算群体稳定性指标（PSI）
    
    参数:
        expected: array-like, 基准期数据（如训练集评分）
        actual: array-like, 监控期数据（如当前生产评分）
        bins: int, 分箱数量
        bin_type: str, 分箱类型：'equal_frequency' 或 'equal_width'
    
    返回:
        psi: float, PSI值
        bin_df: DataFrame, 各箱详细统计
    """
    # 合并数据确定全局最小最大值
    all_data = np.concatenate([expected, actual])
    min_val, max_val = np.min(all_data), np.max(all_data)
    
    # 根据分箱策略生成边界
    if bin_type == 'equal_frequency':
        # 等频分箱：基于基准期数据的分位数
        percentiles = np.linspace(0, 100, bins + 1)
        # 使用 linspace 确保边界覆盖全部范围
        bin_edges = np.percentile(expected, percentiles)
        # 确保边界唯一且覆盖全部数据
        bin_edges[0] = min_val - 1e-6  # 左边界略小于最小值
        bin_edges[-1] = max_val + 1e-6  # 右边界略大于最大值
    elif bin_type == 'equal_width':
        # 等距分箱：基于全局范围的均匀分割
        bin_edges = np.linspace(min_val - 1e-6, max_val + 1e-6, bins + 1)
    else:
        raise ValueError("bin_type 必须是 'equal_frequency' 或 'equal_width'")
    
    # 分箱统计
    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

4.4 实战案例：模型评分稳定性监控

假设某信贷风控模型在2024年1月上线，使用训练集评分作为基准分布，现在需要监控2024年6月的生产评分是否发生漂移。

# 生成模拟数据
np.random.seed(42)
n_train = 10000
n_prod = 8000

# 基准期：训练集评分（假设评分范围300-900）
expected_scores = np.random.normal(loc=650, scale=80, size=n_train)
expected_scores = np.clip(expected_scores, 300, 900)

# 监控期：生产评分（模拟轻度漂移，均值降低）
actual_scores = np.random.normal(loc=630, scale=85, size=n_prod)
actual_scores = np.clip(actual_scores, 300, 900)

# 计算PSI（等频分箱，10箱）
psi_ef, df_ef = calculate_psi(expected_scores, actual_scores, bins=10, bin_type='equal_frequency')
print(f"等频分箱 PSI = {psi_ef:.4f}")

# 计算PSI（等距分箱，10箱）
psi_ew, df_ew = calculate_psi(expected_scores, actual_scores, bins=10, bin_type='equal_width')
print(f"等距分箱 PSI = {psi_ew:.4f}")

# 输出各箱详细统计（等频分箱示例）
print("\n等频分箱各箱统计：")
print(df_ef.to_string(index=False))

# 输出结果示例
等频分箱 PSI = 0.0423
等距分箱 PSI = 0.0387

等频分箱各箱统计：
 bin_start   bin_end  expected_count  actual_count  expected_pct  actual_pct  psi_component
  299.9999  560.0000            1000           1050        0.1000      0.1312       0.008532
  560.0000  600.0000            1000           1100        0.1000      0.1375       0.010214
  600.0000  630.0000            1000           1050        0.1000      0.1312       0.008532
  630.0000  655.0000            1000            950        0.1000      0.1188       0.003174
  655.0000  680.0000            1000            900        0.1000      0.1125       0.001466
  680.0000  705.0000            1000            850        0.1000      0.1062       0.000376
  705.0000  730.0000            1000            800        0.1000      0.1000       0.000000
  730.0000  760.0000            1000            750        0.1000      0.0938       0.000406
  760.0000  800.0000            1000            700        0.1000      0.0875       0.001668
  800.0000  900.0001            1000            800        0.1000      0.1000       0.000000

  结果解读：
  PSI = 0.0423（等频） 和 PSI = 0.0387（等距） 均小于 0.1，表明模型评分分布未发生显著漂移，模型状态稳定。
从各箱统计可见，监控期在低分段（560以下）的占比明显上升（从10%升至13.12%），而高分段（760以上）占比下降。这种趋势虽然尚未触发PSI告警阈值，但已提示模型对低风险客户的区分能力可能正在减弱。
等频分箱 vs 等距分箱：本例中等频分箱的PSI略高于等距分箱，因为等频分箱在数据密集区域（600-700分）提供了更精细的划分，更敏感地捕捉到了分布变化。

4.5 生产环境PSI监控架构

在实际风控系统中，PSI监控通常作为数据漂移哨兵的一部分，以自动化流水线的方式运行。以下架构图展示了从数据采集到告警的完整链路：

4.6 代码封装：生产级PSI监控类

将PSI计算封装为可复用的监控类，便于集成到数据管道中：

class PSIMonitor:
    """生产级PSI监控器"""
    
    def __init__(self, reference_data, bins=10, bin_type='equal_frequency'):
        """
        初始化监控器
        
        参数:
            reference_data: array-like, 基准期数据
            bins: int, 分箱数量
            bin_type: str, 分箱类型
        """
        self.reference = np.array(reference_data)
        self.bins = bins
        self.bin_type = bin_type
        self.bin_edges = None
        self.ref_pct = None
        self._fit()
    
    def _fit(self):
        """基于基准数据计算分箱边界和基准占比"""
        if self.bin_type == 'equal_frequency':
            percentiles = np.linspace(0, 100, self.bins + 1)
            self.bin_edges = np.percentile(self.reference, percentiles)
            self.bin_edges[0] = np.min(self.reference) - 1e-6
            self.bin_edges[-1] = np.max(self.reference) + 1e-6
        elif self.bin_type == 'equal_width':
            min_val, max_val = np.min(self.reference), np.max(self.reference)
            self.bin_edges = np.linspace(min_val - 1e-6, max_val + 1e-6, self.bins + 1)
        
        ref_counts, _ = np.histogram(self.reference, bins=self.bin_edges)
        self.ref_pct = ref_counts / len(self.reference) + 1e-6
    
    def monitor(self, current_data):
        """
        监控当前数据，计算PSI
        
        参数:
            current_data: array-like, 当前监控期数据
        
        返回:
            dict: 包含PSI值、各箱统计、告警级别
        """
        current = np.array(current_data)
        cur_counts, _ = np.histogram(current, bins=self.bin_edges)
        cur_pct = cur_counts / len(current) + 1e-6
        
        psi_components = (cur_pct - self.ref_pct) * np.log(cur_pct / self.ref_pct)
        psi = np.sum(psi_components)
        
        # 告警级别判定
        if psi < 0.1:
            level = "正常"
        elif psi < 0.25:
            level = "关注"
        else:
            level = "告警"
        
        return {
            'psi': psi,
            'level': level,
            'bin_edges': self.bin_edges,
            'ref_pct': self.ref_pct - 1e-6,
            'cur_pct': cur_pct - 1e-6,
            'psi_components': psi_components
        }

# 使用示例
monitor = PSIMonitor(expected_scores, bins=10, bin_type='equal_frequency')
result = monitor.monitor(actual_scores)
print(f"PSI: {result['psi']:.4f}, 告警级别: {result['level']}")

# 输出
PSI: 0.0423, 告警级别: 正常

生产环境最佳实践：

基准期更新：建议每季度或每半年重新计算基准分布，避免因业务自然演变导致频繁误告警。
多维度监控：除了模型评分，还应对每个入模特征单独计算PSI，定位漂移源头。
联动告警：当PSI超过阈值时，自动触发特征重要性分析、模型性能回测和人工复核流程。
日志记录：每次监控结果（PSI值、各箱占比、时间戳）应持久化存储，用于趋势分析和事后回溯。

4.7 常见问题与解决方案

问题	原因	解决方案
PSI计算结果为NaN	某箱占比为0，导致 ln(0) 计算	添加平滑因子 epsilon=1e-6；或合并空箱
等频分箱边界不唯一	数据中存在大量重复值	使用 np.unique 去重后计算分位数；或增加 jitter 噪声
等距分箱出现空箱	数据分布极不均匀	改用等频分箱；或自定义分箱合并稀疏箱
PSI值长期偏高但模型性能正常	业务自然演变导致分布偏移	更新基准期数据；或放宽阈值至0.15~0.2
PSI值偏低但模型性能下降	分箱过粗掩盖了局部漂移	增加分箱数量（如20箱）；或使用分箱内统计检验