3. PSI计算实战：使用Python计算单个特征与模型分数的PSI值

在掌握了PSI的理论公式与分箱逻辑后，本节将进入实战环节。我们将使用Python从零开始实现PSI计算，分别针对单个特征和模型预测分数两种场景进行演示。通过手写代码，你可以深入理解PSI的每一步计算细节，避免对黑箱库函数的依赖。

3.1 计算环境与数据准备

本实战使用Python 3.8+环境，依赖库仅需pandas和numpy。我们构造两组模拟数据：训练集（基准期）和验证集（监控期），分别包含一个特征变量feature_income和模型预测分数model_score。

import pandas as pd
import numpy as np

# 设置随机种子
np.random.seed(42)

# 训练集（基准期）
n_train = 10000
train_income = np.random.lognormal(mean=10.0, sigma=0.5, size=n_train)
train_score = np.random.normal(loc=650, scale=50, size=n_train)
train_score = np.clip(train_score, 300, 850).astype(int)

df_train = pd.DataFrame({
    'feature_income': train_income,
    'model_score': train_score
})

# 验证集（监控期）—— 加入偏移
n_valid = 8000
valid_income = np.random.lognormal(mean=10.2, sigma=0.6, size=n_valid)
valid_score = np.random.normal(loc=620, scale=60, size=n_valid)
valid_score = np.clip(valid_score, 300, 850).astype(int)

df_valid = pd.DataFrame({
    'feature_income': valid_income,
    'model_score': valid_score
})

print(f"训练集样本数: {len(df_train)}, 验证集样本数: {len(df_valid)}")

3.2 核心PSI计算函数

我们编写一个通用的PSI计算函数，支持等距分箱和等频分箱两种方式。函数接收基准数据、监控数据、分箱数及分箱方式，返回PSI值及详细的分箱分布表。

def calculate_psi(base, current, bins=10, bin_type='equal_width'):
    """
    计算PSI（群体稳定性指标）
    
    参数:
        base: 基准期数据（Series）
        current: 监控期数据（Series）
        bins: 分箱数量
        bin_type: 'equal_width' 等距分箱, 'equal_freq' 等频分箱
    
    返回:
        psi_value: PSI值
        psi_df: 包含各分箱详细信息的DataFrame
    """
    # 合并数据确定分箱边界
    all_data = pd.concat([base, current], axis=0)
    
    if bin_type == 'equal_width':
        # 等距分箱
        bin_edges = np.linspace(all_data.min(), all_data.max(), bins + 1)
    elif bin_type == 'equal_freq':
        # 等频分箱（基于基准期）
        bin_edges = np.percentile(base, np.linspace(0, 100, bins + 1))
        # 确保边界唯一
        bin_edges = np.unique(bin_edges)
        if len(bin_edges) < bins + 1:
            # 若分位数重复，退化为等距
            bin_edges = np.linspace(all_data.min(), all_data.max(), bins + 1)
    else:
        raise ValueError("bin_type 必须为 'equal_width' 或 'equal_freq'")
    
    # 分箱
    base_bins = pd.cut(base, bins=bin_edges, include_lowest=True, right=True)
    current_bins = pd.cut(current, bins=bin_edges, include_lowest=True, right=True)
    
    # 计算各箱占比
    base_pct = base_bins.value_counts(normalize=True).sort_index().values
    current_pct = current_bins.value_counts(normalize=True).sort_index().values
    
    # 构建结果DataFrame
    bin_labels = [f"({bin_edges[i]:.2f}, {bin_edges[i+1]:.2f}]" for i in range(len(bin_edges)-1)]
    psi_df = pd.DataFrame({
        'bin_range': bin_labels,
        'base_pct': base_pct,
        'current_pct': current_pct
    })
    
    # 计算各箱PSI贡献
    psi_df['psi_contrib'] = (psi_df['current_pct'] - psi_df['base_pct']) * \
                             np.log(psi_df['current_pct'] / psi_df['base_pct'])
    
    # 处理极端情况：占比为0时用极小值替代
    psi_df['psi_contrib'] = psi_df['psi_contrib'].replace([np.inf, -np.inf], 0).fillna(0)
    
    psi_value = psi_df['psi_contrib'].sum()
    
    return psi_value, psi_df

3.3 实战一：计算单个特征的PSI

以特征feature_income为例，分别使用等距分箱和等频分箱计算PSI，观察不同分箱策略对结果的影响。

# 提取特征数据
base_income = df_train['feature_income']
valid_income = df_valid['feature_income']

# 等距分箱
psi_equal, df_equal = calculate_psi(base_income, valid_income, bins=10, bin_type='equal_width')
print(f"等距分箱 PSI = {psi_equal:.6f}")

# 等频分箱
psi_freq, df_freq = calculate_psi(base_income, valid_income, bins=10, bin_type='equal_freq')
print(f"等频分箱 PSI = {psi_freq:.6f}")

查看等距分箱的详细分布表：

print(df_equal.head(5))

3.4 实战二：计算模型分数的PSI

模型分数是信用评分模型最核心的监控对象。我们计算model_score的PSI，并可视化各分箱的占比变化。

# 提取模型分数数据
base_score = df_train['model_score']
valid_score = df_valid['model_score']

# 使用等距分箱（分数通常使用等距分箱，便于业务解释）
psi_score, df_score = calculate_psi(base_score, valid_score, bins=10, bin_type='equal_width')
print(f"模型分数 PSI = {psi_score:.6f}")

# 查看各分箱详情
print(df_score)

3.5 PSI计算流程可视化

下图展示了PSI计算的完整流程，从数据输入到最终输出PSI值及分箱明细。

3.6 批量计算与自动化

在实际项目中，我们需要同时监控多个特征和模型分数。下面封装一个批量计算函数，返回所有监控项的PSI汇总表。

def batch_psi_calculation(base_df, current_df, feature_list, bins=10, bin_type='equal_width'):
    """
    批量计算多个特征的PSI
    
    参数:
        base_df: 基准期DataFrame
        current_df: 监控期DataFrame
        feature_list: 需要计算PSI的特征名列表
        bins: 分箱数
        bin_type: 分箱方式
    
    返回:
        psi_summary: PSI汇总DataFrame
    """
    results = []
    for feat in feature_list:
        if feat not in base_df.columns or feat not in current_df.columns:
            print(f"警告: 特征 {feat} 不存在，跳过")
            continue
        psi_val, _ = calculate_psi(base_df[feat], current_df[feat], bins=bins, bin_type=bin_type)
        results.append({'feature': feat, 'psi_value': psi_val})
    
    psi_summary = pd.DataFrame(results)
    psi_summary = psi_summary.sort_values('psi_value', ascending=False).reset_index(drop=True)
    return psi_summary

# 批量计算示例
features_to_monitor = ['feature_income', 'model_score']
psi_summary = batch_psi_calculation(df_train, df_valid, features_to_monitor, bins=10, bin_type='equal_freq')
print(psi_summary)

3.7 常见陷阱与避坑指南

• 分箱数选择：分箱过少（<5）会掩盖局部漂移，过多（>20）会导致各箱样本稀疏，PSI不稳定。建议信用评分模型使用10-15箱。
• 零占比处理：当某箱在基准期或监控期占比为0时，log(0)会报错。务必使用极小值（如1e-6）替代，或跳过该箱。
• 分箱边界一致性：监控期的分箱边界必须与基准期完全一致，不能重新计算边界。否则PSI会失去可比性。
• 异常值影响：极端异常值会拉伸分箱边界，导致大部分样本集中在少数箱中。建议先进行截尾处理（如1%和99%分位数截断）。
• 业务含义验证：PSI数值低不一定代表模型稳定，需结合业务逻辑检查分箱分布是否合理。例如，分数整体右移但PSI很低，可能是分箱边界设置不当。