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

本章将使用 Python 完整实现 PSI 的计算流程，包括分箱函数、PSI 计算函数封装以及结果解读。通过实战代码，你将掌握从原始分数分布到 PSI 指标输出的全链路实现。

4.1 分箱函数编写

PSI 计算的第一步是将连续分数划分为离散区间（分箱）。分箱方式直接影响 PSI 的稳定性。常用的分箱方法包括等距分箱、等频分箱和自定义分箱。

分箱原则： 分箱数量通常建议在 5-20 之间，过少会丢失分布细节，过多则可能放大随机波动。对于评分卡模型，常使用 10 等分或 20 等分。

以下代码实现了三种分箱函数：

import pandas as pd
import numpy as np

def equal_width_binning(series, bins=10):
    """
    等距分箱：将数据按数值范围均匀划分为 bins 个区间
    """
    # 使用 pd.cut 进行等距分箱
    labels = [f'Bin_{i+1}' for i in range(bins)]
    binned = pd.cut(series, bins=bins, labels=labels, include_lowest=True)
    return binned

def equal_freq_binning(series, bins=10):
    """
    等频分箱：将数据按样本数量均匀划分为 bins 个区间
    """
    # 使用 pd.qcut 进行等频分箱
    labels = [f'Bin_{i+1}' for i in range(bins)]
    try:
        binned = pd.qcut(series, q=bins, labels=labels, duplicates='drop')
    except ValueError:
        # 当分位数重复时，使用 rank 方法近似
        rank_series = series.rank(method='first')
        binned = pd.qcut(rank_series, q=bins, labels=labels, duplicates='drop')
    return binned

def custom_binning(series, cut_points):
    """
    自定义分箱：根据指定的切分点进行分箱
    cut_points: 列表，如 [0, 300, 500, 700, 1000]
    """
    labels = [f'Bin_{i+1}' for i in range(len(cut_points)-1)]
    binned = pd.cut(series, bins=cut_points, labels=labels, include_lowest=True)
    return binned
分箱函数对比：
  • 等距分箱： 适用于分数分布均匀的场景，但若数据存在极端值，会导致部分区间样本过少。
  • 等频分箱： 保证每个区间样本量大致相等，对极端值不敏感，是评分卡 PSI 计算的首选。
  • 自定义分箱： 适用于已知业务切分点（如评分卡阈值）的场景，可保持与业务规则一致。

4.2 PSI 计算函数封装

基于分箱结果，我们封装一个通用的 PSI 计算函数。该函数接受实际分布和预期分布的分箱计数，输出每个分箱的贡献度和整体 PSI 值。

def calculate_psi(actual, expected, bins=10, binning_method='equal_freq'):
    """
    计算 PSI（群体稳定性指标）
    
    参数:
        actual: 实际分布（当前期）的分数 Series
        expected: 预期分布（基准期）的分数 Series
        bins: 分箱数量
        binning_method: 分箱方法，可选 'equal_freq', 'equal_width', 'custom'
    
    返回:
        psi_value: 整体 PSI 值
        psi_detail: 每个分箱的详细贡献度 DataFrame
    """
    # 1. 基于预期分布确定分箱边界
    if binning_method == 'equal_freq':
        # 使用预期分布的分位数作为切分点
        quantiles = np.linspace(0, 1, bins + 1)
        cut_points = expected.quantile(quantiles).values
        # 处理重复切分点
        cut_points = np.unique(cut_points)
        if len(cut_points) < 2:
            raise ValueError("分箱切分点不足，请减少分箱数量或检查数据分布")
        labels = [f'Bin_{i+1}' for i in range(len(cut_points)-1)]
        actual_binned = pd.cut(actual, bins=cut_points, labels=labels, include_lowest=True)
        expected_binned = pd.cut(expected, bins=cut_points, labels=labels, include_lowest=True)
    elif binning_method == 'equal_width':
        # 使用预期分布的最小最大值作为边界
        min_val, max_val = expected.min(), expected.max()
        cut_points = np.linspace(min_val, max_val, bins + 1)
        labels = [f'Bin_{i+1}' for i in range(bins)]
        actual_binned = pd.cut(actual, bins=cut_points, labels=labels, include_lowest=True)
        expected_binned = pd.cut(expected, bins=cut_points, labels=labels, include_lowest=True)
    else:
        raise ValueError("不支持的分箱方法，请使用 'equal_freq' 或 'equal_width'")
    
    # 2. 计算每个分箱的样本占比
    actual_counts = actual_binned.value_counts().sort_index()
    expected_counts = expected_binned.value_counts().sort_index()
    
    actual_pct = actual_counts / actual_counts.sum()
    expected_pct = expected_counts / expected_counts.sum()
    
    # 3. 构建结果 DataFrame
    psi_detail = pd.DataFrame({
        'Bin': actual_pct.index,
        'Actual_Pct': actual_pct.values,
        'Expected_Pct': expected_pct.values
    })
    
    # 4. 计算每个分箱的 PSI 贡献度
    # 处理占比为 0 的情况：用极小值替代避免除零
    epsilon = 1e-10
    psi_detail['Actual_Pct_Adj'] = psi_detail['Actual_Pct'].replace(0, epsilon)
    psi_detail['Expected_Pct_Adj'] = psi_detail['Expected_Pct'].replace(0, epsilon)
    
    psi_detail['PSI_Contrib'] = (psi_detail['Actual_Pct_Adj'] - psi_detail['Expected_Pct_Adj']) * \
                                 np.log(psi_detail['Actual_Pct_Adj'] / psi_detail['Expected_Pct_Adj'])
    
    # 5. 计算整体 PSI
    psi_value = psi_detail['PSI_Contrib'].sum()
    
    return psi_value, psi_detail[['Bin', 'Actual_Pct', 'Expected_Pct', 'PSI_Contrib']]
注意事项：
  • 当某个分箱在预期分布或实际分布中占比为 0 时，会导致对数计算错误。代码中使用极小值 epsilon 替代，避免除零。
  • 分箱边界应基于预期分布（基准期）确定，以保证前后两期使用相同的评分区间。
  • 若实际分布中出现预期分布未覆盖的分数区间，这些样本会被归入最近的分箱或标记为缺失，需在业务层面解释。

4.3 完整计算示例

以下示例演示如何使用上述函数计算评分卡模型的 PSI：

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

# 基准期评分（预期分布）
expected_scores = np.random.normal(loc=600, scale=100, size=n_samples)
expected_scores = np.clip(expected_scores, 200, 1000).astype(int)

# 当前期评分（实际分布）- 模拟分布偏移
actual_scores = np.random.normal(loc=580, scale=110, size=n_samples)
actual_scores = np.clip(actual_scores, 200, 1000).astype(int)

# 计算 PSI
psi_value, psi_detail = calculate_psi(
    actual=pd.Series(actual_scores, name='actual'),
    expected=pd.Series(expected_scores, name='expected'),
    bins=10,
    binning_method='equal_freq'
)

print(f"整体 PSI 值: {psi_value:.6f}")
print("\n分箱详细贡献度:")
print(psi_detail.to_string(index=False))

输出结果示例：

整体 PSI 值: 0.023847

分箱详细贡献度:
    Bin  Actual_Pct  Expected_Pct  PSI_Contrib
  Bin_1     0.0985        0.1000     0.000023
  Bin_2     0.1012        0.1000     0.000014
  Bin_3     0.0978        0.1000     0.000049
  Bin_4     0.1034        0.1000     0.000113
  Bin_5     0.0991        0.1000     0.000008
  Bin_6     0.0965        0.1000     0.000126
  Bin_7     0.1028        0.1000     0.000078
  Bin_8     0.0989        0.1000     0.000012
  Bin_9     0.1045        0.1000     0.000198
 Bin_10     0.0973        0.1000     0.000073

4.4 PSI 计算流程架构

下图展示了 PSI 计算的完整流程，从原始分数输入到最终稳定性判断：

PSI 计算流程架构图 基准期评分 当前期评分 分箱处理 等频分箱 / 等距分箱 占比计算 Actual_Pct vs Expected_Pct PSI 贡献度计算 (A-E) × ln(A/E) 整体 PSI 值 Σ PSI_Contrib 稳定性判断 PSI < 0.1 稳定 分箱注意事项 • 分箱数量 5-20 • 基于基准期确定边界 • 处理占比为 0 的情况 • 保持分箱一致性

4.5 结果解读

PSI 计算结果需要结合业务场景进行解读。以下是 PSI 值的通用判断标准：

PSI 范围 稳定性等级 业务含义 建议行动
PSI < 0.1 稳定 模型评分分布无明显偏移 无需调整，持续监控
0.1 ≤ PSI < 0.25 轻度偏移 评分分布出现轻微变化 分析偏移原因，加强监控频率
PSI ≥ 0.25 严重偏移 评分分布发生显著变化 立即排查原因，考虑模型重建或校准
分箱贡献度分析：

除了整体 PSI 值，每个分箱的贡献度（PSI_Contrib）同样重要。贡献度较大的分箱指示了分布偏移的主要区域。例如，若 Bin_9 的贡献度显著高于其他分箱，说明高分段的评分分布发生了较大变化，需要重点关注该客群的风险特征是否改变。

4.6 函数封装与复用

将 PSI 计算封装为可复用的模块，便于在模型监控系统中集成：

class PSICalculator:
    """
    PSI 计算器类，支持批量计算和结果存储
    """
    def __init__(self, bins=10, binning_method='equal_freq'):
        self.bins = bins
        self.binning_method = binning_method
        self.history = []
    
    def compute(self, actual, expected, label=''):
        """
        计算单次 PSI
        """
        psi_value, psi_detail = calculate_psi(
            actual, expected, 
            bins=self.bins, 
            binning_method=self.binning_method
        )
        result = {
            'label': label,
            'psi_value': psi_value,
            'psi_detail': psi_detail
        }
        self.history.append(result)
        return result
    
    def get_summary(self):
        """
        获取历史计算摘要
        """
        summary = pd.DataFrame([
            {'Label': r['label'], 'PSI': r['psi_value']}
            for r in self.history
        ])
        return summary
    
    def plot_psi_trend(self):
        """
        绘制 PSI 趋势图（需 matplotlib 支持）
        """
        # 实际项目中可集成绘图功能
        pass

# 使用示例
calculator = PSICalculator(bins=10, binning_method='equal_freq')

# 计算月度 PSI
result_jan = calculator.compute(
    actual=pd.Series(actual_scores_jan),
    expected=pd.Series(expected_scores),
    label='2024-01'
)
result_feb = calculator.compute(
    actual=pd.Series(actual_scores_feb),
    expected=pd.Series(expected_scores),
    label='2024-02'
)

# 查看摘要
print(calculator.get_summary())
生产环境建议：
  • 将 PSI 计算封装为独立的微服务或 API，支持定时任务自动计算。
  • 存储每次计算的详细分箱数据，便于回溯分析。
  • 设置 PSI 阈值告警，当 PSI 超过 0.1 或 0.25 时自动触发通知。
  • 结合其他稳定性指标（如特征 PSI、模型 AUC 趋势）综合判断模型健康状况。

4.7 常见问题与避坑

在实际使用 PSI 计算时，以下问题需要特别注意：

问题 原因 解决方案
PSI 计算结果为 NaN 分箱中存在占比为 0 的情况，导致对数计算无效 使用极小值 epsilon 替代 0，或合并样本过少的分箱
PSI 值异常大（>1） 分箱边界不合理，或两期分布差异极大 检查分箱方法，确认数据是否存在异常值或数据错误
不同分箱方法结果差异大 等距分箱对极端值敏感，等频分箱更稳健 优先使用等频分箱，或根据业务场景选择合适方法
样本量不足导致分箱不稳定 分箱数量过多，部分区间样本极少 减少分箱数量，或使用自适应分箱方法
← 上一章 📖 返回目录 下一章 →