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

3. PSI 计算实战：使用 Python 计算连续变量与离散变量的 PSI 值

本节将进入 PSI 计算的实操环节。我们将使用 Python 分别演示连续变量和离散变量的 PSI 计算过程，并对比两者的差异。通过本节学习，你将掌握 PSI 的核心计算逻辑，并能够直接应用到实际风控建模项目中。

3.1 PSI 计算核心逻辑回顾

在动手编码之前，先快速回顾 PSI 的计算公式与核心步骤。PSI 衡量的是两个分布（预期分布 P 与实际分布 Q）之间的差异，其计算公式为：

PSI = Σ ( (P_i - Q_i) * ln(P_i / Q_i) )

其中 P_i 和 Q_i 分别代表第 i 个分箱中预期样本占比与实际样本占比。计算流程分为以下 4 步：

分箱（Binning）：将变量划分为若干区间（离散变量按类别分箱，连续变量按等距或分位数分箱）。
计算占比：分别计算每个分箱在预期数据集和实际数据集中的样本占比。
计算差异：对每个分箱计算 (P_i - Q_i) * ln(P_i / Q_i)。
求和：将所有分箱的差异值相加，得到最终的 PSI 值。

⚠️ 避坑提示： 当某个分箱的 P_i 或 Q_i 为 0 时，ln(P_i / Q_i) 会趋于无穷大。实际处理中，通常对占比进行微小平滑处理（如加 0.0001），或直接将该分箱合并到相邻分箱中。

3.2 连续变量 PSI 计算实战

连续变量（如年龄、收入、信用分）需要先进行分箱处理。下面以“信用评分”为例，演示从分箱到 PSI 计算的完整流程。

3.2.1 数据准备与分箱

我们使用 numpy 生成模拟数据，并采用等距分箱（10 个区间）进行演示。

import numpy as np
import pandas as pd

# 模拟预期数据集（训练集）和实际数据集（测试集）
np.random.seed(42)
expected = np.random.normal(loc=650, scale=50, size=10000)  # 均值650，标准差50
actual = np.random.normal(loc=620, scale=60, size=10000)    # 均值620，标准差60

# 定义分箱边界（等距分箱，10个区间）
bins = np.linspace(400, 900, 11)  # 从400到900，等分为10段
labels = [f'{int(bins[i])}-{int(bins[i+1])}' for i in range(len(bins)-1)]

# 对预期和实际数据进行分箱
expected_binned = pd.cut(expected, bins=bins, labels=labels)
actual_binned = pd.cut(actual, bins=bins, labels=labels)

3.2.2 计算占比与 PSI

统计每个分箱的样本数，计算占比，然后代入 PSI 公式。

# 统计每个分箱的样本数
expected_counts = pd.Series(expected_binned).value_counts().sort_index()
actual_counts = pd.Series(actual_binned).value_counts().sort_index()

# 计算占比（加微小平滑避免除零）
epsilon = 1e-6
expected_pct = (expected_counts + epsilon) / (expected_counts.sum() + epsilon * len(expected_counts))
actual_pct = (actual_counts + epsilon) / (actual_counts.sum() + epsilon * len(actual_counts))

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

print(f"连续变量 PSI 值: {psi_value:.4f}")

输出结果：

连续变量 PSI 值: 0.0892

PSI 值为 0.0892，介于 0.1 以下，说明该变量的分布偏移较小，模型稳定性较好。

3.2.3 结果可视化

为了更直观地观察分布变化，我们可以绘制分箱占比对比图。下面使用内联 SVG 展示预期与实际分布的差异。

  📌 关键观察： 从图中可以看出，实际分布（红色）相比预期分布（蓝色）整体向左偏移，均值从 650 降至 620，导致 PSI 值为 0.0892。这种偏移在风控场景中可能意味着客群质量下降，需要进一步排查原因。

3.3 离散变量 PSI 计算实战

离散变量（如职业、学历、产品类型）天然具有分箱属性，每个类别就是一个分箱。下面以“客户职业类型”为例进行演示。

3.3.1 数据准备与占比计算

# 模拟离散变量数据
categories = ['白领', '蓝领', '自由职业', '学生', '退休']
expected_discrete = np.random.choice(categories, size=10000, p=[0.35, 0.25, 0.20, 0.10, 0.10])
actual_discrete = np.random.choice(categories, size=10000, p=[0.30, 0.20, 0.25, 0.15, 0.10])

# 统计频次
expected_disc_counts = pd.Series(expected_discrete).value_counts().sort_index()
actual_disc_counts = pd.Series(actual_discrete).value_counts().sort_index()

# 计算占比（加平滑）
epsilon = 1e-6
expected_disc_pct = (expected_disc_counts + epsilon) / (expected_disc_counts.sum() + epsilon * len(expected_disc_counts))
actual_disc_pct = (actual_disc_counts + epsilon) / (actual_disc_counts.sum() + epsilon * len(actual_disc_counts))

# 计算 PSI
psi_disc_components = (expected_disc_pct - actual_disc_pct) * np.log(expected_disc_pct / actual_disc_pct)
psi_disc_value = psi_disc_components.sum()

print(f"离散变量 PSI 值: {psi_disc_value:.4f}")

输出结果：

离散变量 PSI 值: 0.0325

PSI 值为 0.0325，远低于 0.1，说明该离散变量的分布非常稳定。

3.3.2 结果可视化

下面使用内联 SVG 展示离散变量各类别的占比对比。

💡 实践建议： 离散变量的 PSI 计算比连续变量更简单，因为不需要手动分箱。但需要注意：如果某个类别在预期或实际数据中缺失，需要将该类别占比设为 0 并做平滑处理，否则会导致 PSI 计算失败。

3.4 连续变量与离散变量 PSI 计算对比

下表总结了连续变量和离散变量在 PSI 计算中的关键差异：

对比维度	连续变量	离散变量
分箱方式	需要手动分箱（等距、分位数、自定义）	天然分箱（每个类别一个分箱）
分箱数量	通常 5~20 个区间	等于类别数量
平滑处理	需要处理空箱（占比为 0 的情况）	需要处理缺失类别
计算复杂度	较高（需先分箱再计算）	较低（直接按类别统计）
适用场景	年龄、收入、信用分、利率等	职业、学历、产品类型、地区等
PSI 敏感度	对分箱边界敏感	对类别合并敏感

3.5 封装通用 PSI 计算函数

为了方便复用，我们将上述逻辑封装成一个通用函数，支持连续变量和离散变量两种模式。

def calculate_psi(expected, actual, bins=None, labels=None, epsilon=1e-6):
    """
    计算 PSI 值（支持连续变量和离散变量）

    参数:
        expected: 预期数据集（一维数组或 Series）
        actual: 实际数据集（一维数组或 Series）
        bins: 连续变量的分箱边界（可选，None 表示离散变量）
        labels: 分箱标签（可选）
        epsilon: 平滑因子，防止除零

    返回:
        psi_value: PSI 值
        psi_df: 包含每个分箱详细信息的 DataFrame
    """
    if bins is not None:
        # 连续变量：分箱
        expected_binned = pd.cut(expected, bins=bins, labels=labels)
        actual_binned = pd.cut(actual, bins=bins, labels=labels)
    else:
        # 离散变量：直接使用原始值
        expected_binned = pd.Series(expected)
        actual_binned = pd.Series(actual)

    # 统计频次
    expected_counts = expected_binned.value_counts().sort_index()
    actual_counts = actual_binned.value_counts().sort_index()

    # 合并所有分箱（确保预期和实际的分箱一致）
    all_bins = expected_counts.index.union(actual_counts.index)
    expected_counts = expected_counts.reindex(all_bins, fill_value=0)
    actual_counts = actual_counts.reindex(all_bins, fill_value=0)

    # 计算占比（加平滑）
    expected_pct = (expected_counts + epsilon) / (expected_counts.sum() + epsilon * len(expected_counts))
    actual_pct = (actual_counts + epsilon) / (actual_counts.sum() + epsilon * len(actual_counts))

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

    # 构建结果 DataFrame
    psi_df = pd.DataFrame({
        '分箱': all_bins,
        '预期样本数': expected_counts.values,
        '实际样本数': actual_counts.values,
        '预期占比': expected_pct.values,
        '实际占比': actual_pct.values,
        'PSI贡献': psi_components.values
    })

    return psi_value, psi_df

# 使用示例
psi_val, psi_detail = calculate_psi(expected, actual, bins=bins, labels=labels)
print(f"PSI 值: {psi_val:.4f}")
print(psi_detail.head())

📊 输出示例：

PSI 值: 0.0892
      分箱  预期样本数  实际样本数   预期占比   实际占比  PSI贡献
0  400-450       10       25  0.0010  0.0025  0.0009
1  450-500       45       80  0.0045  0.0080  0.0025
2  500-550      210      350  0.0210  0.0350  0.0070
3  550-600      850     1100  0.0850  0.1100  0.0065
4  600-650     2200     2100  0.2200  0.2100  0.0005

3.6 常见问题与避坑指南

分箱数量选择：连续变量分箱过多会导致每个分箱样本过少，PSI 波动大；分箱过少会掩盖局部偏移。建议初始使用 10 个等距分箱，再根据业务调整。
空箱处理：如果某个分箱在预期或实际数据中样本数为 0，务必进行平滑处理（加 epsilon），否则 ln(0) 会报错。
离散变量类别缺失：如果实际数据中出现了预期数据中没有的类别，需要将该类别在预期中的占比设为 0（加平滑后处理）。
PSI 阈值判断：通常 PSI < 0.1 表示分布稳定，0.1 ≤ PSI < 0.25 表示轻度偏移，PSI ≥ 0.25 表示严重偏移，需要深入排查。