4. PSI计算实战：使用Python计算单个因子的PSI值（分箱法、密度法）

PSI（Population Stability Index，群体稳定性指标）是量化风控中衡量因子分布随时间漂移程度的核心指标。本节将深入讲解PSI的两种主流计算方法——分箱法与密度法，并提供可直接复用的Python代码实现。

4.1 PSI计算原理回顾

PSI通过比较因子在基准期（如建模样本）与监测期（如最新线上数据）的分布差异，量化漂移程度。其核心公式为：

PSI = Σ (P_i_actual - P_i_expected) * ln(P_i_actual / P_i_expected)

其中：

• P_i_expected：基准期第 i 个分箱的样本占比
• P_i_actual：监测期第 i 个分箱的样本占比

4.2 方法一：分箱法（等频分箱）

分箱法是最常用的PSI计算方式，适用于连续型因子。其核心步骤为：基于基准期数据确定分箱边界，然后将监测期数据映射到相同的分箱中，最后计算各箱占比并代入PSI公式。

4.2.1 分箱法计算流程

1. 确定分箱数：通常选择 10 箱（等频分箱），也可根据样本量调整。
2. 计算基准期分箱边界：使用 pd.qcut() 对基准期因子值进行等频分箱。
3. 映射监测期数据：将监测期因子值按基准期边界进行分箱。
4. 计算各箱占比：分别计算基准期和监测期每个分箱的样本占比。
5. 计算PSI：逐箱计算 (实际占比 - 期望占比) * ln(实际占比 / 期望占比)，并求和。

4.2.2 分箱法Python实现

import pandas as pd
import numpy as np

def psi_binning(expected, actual, bins=10):
    """
    分箱法计算PSI
    :param expected: 基准期因子值 (Series)
    :param actual: 监测期因子值 (Series)
    :param bins: 分箱数，默认10
    :return: PSI值
    """
    # 1. 基于基准期数据确定分箱边界（等频分箱）
    expected_bins = pd.qcut(expected, q=bins, duplicates='drop', retbins=True)[1]
    
    # 2. 将基准期和监测期数据按相同边界分箱
    expected_cut = pd.cut(expected, bins=expected_bins, include_lowest=True)
    actual_cut = pd.cut(actual, bins=expected_bins, include_lowest=True)
    
    # 3. 计算各箱占比
    expected_pct = expected_cut.value_counts(normalize=True).sort_index()
    actual_pct = actual_cut.value_counts(normalize=True).sort_index()
    
    # 4. 合并并处理空箱（占比为0时，用极小值替代避免除零）
    df = pd.DataFrame({'expected': expected_pct, 'actual': actual_pct})
    df = df.fillna(0.0001)  # 避免ln(0)
    df['actual'] = df['actual'].replace(0, 0.0001)
    
    # 5. 计算PSI
    df['psi_component'] = (df['actual'] - df['expected']) * np.log(df['actual'] / df['expected'])
    psi = df['psi_component'].sum()
    
    return psi

# 示例用法
# psi_value = psi_binning(train_df['feature_x'], online_df['feature_x'], bins=10)
# print(f"分箱法PSI: {psi_value:.4f}")

4.3 方法二：密度法（核密度估计）

密度法适用于样本量较大或分箱边界难以确定的场景。它通过核密度估计（KDE）拟合因子分布，然后对密度曲线进行离散化采样，再计算PSI。

4.3.1 密度法计算流程

1. 拟合密度曲线：分别对基准期和监测期因子值进行KDE拟合。
2. 定义采样点：在因子值范围内均匀取 N 个点（通常 N=1000）。
3. 计算密度值：在每个采样点上计算两个分布的密度值。
4. 归一化：将密度值归一化为概率（占比）。
5. 计算PSI：使用与分箱法相同的公式，逐点计算并求和。

4.3.2 密度法Python实现

from scipy.stats import gaussian_kde
import numpy as np

def psi_density(expected, actual, n_points=1000):
    """
    密度法计算PSI
    :param expected: 基准期因子值 (array-like)
    :param actual: 监测期因子值 (array-like)
    :param n_points: 采样点数，默认1000
    :return: PSI值
    """
    # 1. 拟合KDE
    kde_expected = gaussian_kde(expected)
    kde_actual = gaussian_kde(actual)
    
    # 2. 定义采样范围（取两个分布并集的1%和99%分位数）
    combined = np.concatenate([expected, actual])
    x_min = np.percentile(combined, 1)
    x_max = np.percentile(combined, 99)
    x = np.linspace(x_min, x_max, n_points)
    
    # 3. 计算密度值
    density_expected = kde_expected.evaluate(x)
    density_actual = kde_actual.evaluate(x)
    
    # 4. 归一化为概率
    prob_expected = density_expected / density_expected.sum()
    prob_actual = density_actual / density_actual.sum()
    
    # 5. 处理零值并计算PSI
    prob_expected = np.clip(prob_expected, 1e-10, None)
    prob_actual = np.clip(prob_actual, 1e-10, None)
    
    psi = np.sum((prob_actual - prob_expected) * np.log(prob_actual / prob_expected))
    
    return psi

# 示例用法
# psi_value = psi_density(train_df['feature_x'].values, online_df['feature_x'].values)
# print(f"密度法PSI: {psi_value:.4f}")

4.4 两种方法对比

4.5 PSI计算流程总览

下图展示了从因子数据输入到PSI输出的完整计算链路，涵盖分箱法与密度法两条路径：

4.6 实战建议与避坑指南

• 分箱数选择：样本量大于10万时，可尝试20箱；样本量小于5000时，建议降至5箱，避免空箱过多。
• 空箱处理：分箱法中若监测期某箱占比为0，建议用 0.0001 替代，而非直接丢弃该箱，否则会低估PSI。
• 密度法带宽：scipy 默认使用 Scott 规则，若分布波动剧烈，可尝试 gaussian_kde(expected, bw_method='silverman') 获得更平滑的曲线。
• 性能优化：对全量因子计算PSI时，建议使用 apply 或向量化操作，避免逐因子循环。

4.7 完整代码封装

以下代码将两种方法封装为统一接口，便于批量计算：

def calculate_psi(expected, actual, method='binning', bins=10, n_points=1000):
    """
    统一PSI计算接口
    :param method: 'binning' 或 'density'
    """
    if method == 'binning':
        return psi_binning(expected, actual, bins)
    elif method == 'density':
        return psi_density(expected, actual, n_points)
    else:
        raise ValueError("method must be 'binning' or 'density'")

# 批量计算示例
# feature_list = ['age', 'income', 'credit_limit']
# psi_results = {}
# for feat in feature_list:
#     psi_results[feat] = calculate_psi(
#         train_df[feat], online_df[feat], method='binning', bins=10
#     )
# print(psi_results)