4. PSI计算实战：使用Python计算两个分布之间的PSI值，代码实现与结果解读

本节将进入PSI计算的实战环节。我们将使用Python从零实现PSI计算函数，并基于一个模拟的评分卡分数分布案例，演示如何计算两个分布（预期分布 vs. 实际分布）之间的PSI值，并对计算结果进行专业解读。

4.1 PSI计算核心逻辑

PSI的计算基于分箱（Binning）思想，核心步骤如下：

  1. 分箱：将评分或概率区间划分为K个等宽或等频的箱子。
  2. 计算占比：分别计算每个箱子中，预期样本数占总预期样本数的比例（\(P_i\)），以及实际样本数占总实际样本数的比例（\(Q_i\)）。
  3. 计算每个箱子的PSI贡献：\((P_i - Q_i) \times \ln(P_i / Q_i)\)。
  4. 求和：将所有箱子的贡献值相加，得到最终的PSI值。
关键公式：
\(PSI = \sum_{i=1}^{K} (P_i - Q_i) \times \ln\left(\frac{P_i}{Q_i}\right)\)
其中，\(P_i\) 为预期分布在第i个箱子的占比，\(Q_i\) 为实际分布在第i个箱子的占比。

4.2 Python代码实现

以下代码定义了一个通用的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'（等频）或'width'（等宽）。

    返回:
        float: PSI值。
        pd.DataFrame: 包含每个箱子的详细计算过程。
    """
    # 1. 合并数据确定分箱边界
    all_scores = np.concatenate([expected, actual])
    
    if bucket_type == 'quantile':
        # 等频分箱：基于预期分布的分位数
        percentiles = np.linspace(0, 100, bins + 1)[1:-1]
        bin_edges = np.percentile(expected, percentiles)
        # 添加最小和最大边界
        bin_edges = np.concatenate([[-np.inf], bin_edges, [np.inf]])
    else:
        # 等宽分箱
        min_val, max_val = np.min(all_scores), np.max(all_scores)
        bin_edges = np.linspace(min_val, max_val, bins + 1)
        bin_edges[0] = -np.inf
        bin_edges[-1] = np.inf

    # 2. 分箱并计算占比
    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)

    # 3. 处理零值：将0替换为一个极小值，避免除零或log(0)
    epsilon = 1e-6
    expected_pct = np.where(expected_pct == 0, epsilon, expected_pct)
    actual_pct = np.where(actual_pct == 0, epsilon, actual_pct)

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

    # 5. 构建结果DataFrame
    bin_labels = [f'Bin_{i+1}' for i in range(bins)]
    result_df = pd.DataFrame({
        'Bin': bin_labels,
        'Expected_Count': expected_counts,
        'Actual_Count': actual_counts,
        'Expected_Pct': expected_pct,
        'Actual_Pct': actual_pct,
        'PSI_Component': psi_components
    })

    return psi_value, result_df
代码要点：
  • 分箱方式选择：等频分箱（quantile）能保证每个箱子样本量大致相等，更稳定；等宽分箱（width）在分数分布极不均匀时可能导致空箱。
  • 零值处理：使用 epsilon = 1e-6 替代0，避免 log(0) 或除零错误。
  • 边界处理：使用 -np.inf 和 np.inf 确保所有数据点都能落入某个箱子。

4.3 实战案例：评分卡分数漂移监控

假设我们有一个信用评分卡模型，分数范围300~850。以下是建模时的训练集分数（预期分布）和当前月份的线上分数（实际分布）的模拟数据。

# 模拟数据
np.random.seed(42)
expected_scores = np.random.normal(loc=650, scale=50, size=10000)
actual_scores = np.random.normal(loc=620, scale=60, size=10000)  # 均值下降，方差增大

# 计算PSI
psi_value, psi_df = calculate_psi(expected_scores, actual_scores, bins=10, bucket_type='quantile')

print(f"总体PSI值: {psi_value:.4f}")
print("\n各箱子详细计算过程:")
print(psi_df.to_string(index=False))

输出结果示例：

总体PSI值: 0.0892

各箱子详细计算过程:
    Bin  Expected_Count  Actual_Count  Expected_Pct  Actual_Pct  PSI_Component
  Bin_1            1000           1350         0.1000       0.1350         0.0102
  Bin_2            1000           1200         0.1000       0.1200         0.0036
  Bin_3            1000           1100         0.1000       0.1100         0.0009
  Bin_4            1000           1050         0.1000       0.1050         0.0002
  Bin_5            1000           1000         0.1000       0.1000         0.0000
  Bin_6            1000            950         0.1000       0.0950         0.0003
  Bin_7            1000            900         0.1000       0.0900         0.0011
  Bin_8            1000            800         0.1000       0.0800         0.0045
  Bin_9            1000            750         0.1000       0.0750         0.0072
 Bin_10            1000            900         0.1000       0.0900         0.0011

4.4 结果解读与漂移判定

根据行业通用标准，PSI值的解读如下：

PSI值范围 漂移程度 建议行动
PSI < 0.1 无显著漂移 无需干预，持续监控
0.1 ≤ PSI < 0.25 轻度漂移 需关注，排查特征或客群变化
PSI ≥ 0.25 严重漂移 立即告警，启动模型重训或策略调整

在本案例中，PSI值为0.0892，小于0.1，属于无显著漂移。但从各箱子的PSI贡献来看，Bin_1（低分段）和Bin_9（高分段）的贡献值相对较高，说明实际分布的低分客群占比增加，高分客群占比减少，模型评分整体有下移趋势，需要持续关注。

实战避坑：
  • 分箱数量：bins=10是行业默认值，但若样本量较小（如<1000），建议减少至5~6箱，避免空箱导致PSI失真。
  • 分箱方式：等频分箱比等宽分箱更稳健，尤其在分数分布呈偏态时。
  • 零值处理：务必处理空箱，否则PSI计算会报错或产生无穷大值。
  • 阈值调整：不同业务场景可微调PSI阈值，例如高风险信贷场景可将告警阈值设为0.15。

4.5 PSI计算流程可视化

下图展示了从原始分数分布到最终PSI值的完整计算流程，帮助理解数据流转与核心步骤。

PSI计算流程图 预期分布 (P) 实际分布 (Q) 分箱 (等频/等宽, bins=10) 计算各箱占比 (P_i, Q_i) 计算各箱PSI贡献值 求和 → 最终PSI值 PSI = Σ(P_i - Q_i) * ln(P_i/Q_i) 阈值: <0.1正常, 0.1~0.25关注, ≥0.25告警

4.6 批量监控与自动化脚本

在实际生产环境中，通常需要按天/周/月批量计算PSI，并自动触发告警。以下是一个简化的批量监控脚本框架：

import pandas as pd
from datetime import datetime, timedelta

def batch_psi_monitor(historical_data, current_data, date_col='score_date', score_col='score', bins=10):
    """
    批量计算每日PSI，并与基准分布对比。
    
    参数:
        historical_data (pd.DataFrame): 历史基准数据（含日期和分数列）。
        current_data (pd.DataFrame): 当前监控期数据。
        date_col (str): 日期列名。
        score_col (str): 分数列名。
        bins (int): 分箱数。
    
    返回:
        pd.DataFrame: 每日PSI监控结果。
    """
    # 以历史数据为基准分布
    expected = historical_data[score_col].values
    
    results = []
    for date, group in current_data.groupby(date_col):
        actual = group[score_col].values
        psi, _ = calculate_psi(expected, actual, bins=bins)
        results.append({'date': date, 'psi': psi})
    
    result_df = pd.DataFrame(results)
    result_df['alert_level'] = result_df['psi'].apply(
        lambda x: '正常' if x < 0.1 else ('关注' if x < 0.25 else '告警')
    )
    return result_df

# 使用示例
# daily_psi = batch_psi_monitor(train_data, online_data)
# print(daily_psi[daily_psi['alert_level'] == '告警'])
← 上一章 📖 返回目录 下一章 →