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

本节将使用Python完整实现PSI（群体稳定性指标）的计算，涵盖数据准备、分箱处理、pandas/numpy核心操作以及结果解读。通过实战代码，你将掌握PSI的自动化监控流程，并能够快速定位模型或特征分布偏移的根源。

4.1 PSI计算核心逻辑回顾

PSI衡量的是预期分布（通常为建模时的训练集或验证集）与实际分布（当前监控期）之间的差异。其计算公式为：

PSI = Σ (实际占比_i - 预期占比_i) * ln(实际占比_i / 预期占比_i)

其中，i 代表第 i 个分箱。计算流程包含三个关键步骤：

1. 分箱：将连续型分数或概率划分为离散区间（通常为10-20箱）。
2. 计算占比：统计每个分箱中样本数占总样本数的比例。
3. 求和：对每个分箱计算上述公式并累加。

4.2 完整Python代码实现

以下代码使用pandas进行数据分组与聚合，numpy进行向量化运算，避免显式循环，提升计算效率。

import pandas as pd
import numpy as np

def calculate_psi(expected, actual, bins=10, bucket_type='quantile'):
    """
    计算PSI（群体稳定性指标）
    
    参数:
        expected (pd.Series): 预期分布（如训练集分数）
        actual (pd.Series): 实际分布（如当前监控期分数）
        bins (int): 分箱数量，默认10
        bucket_type (str): 分箱方式，'quantile' 等频分箱，'width' 等宽分箱
    
    返回:
        psi_value (float): PSI值
        psi_df (pd.DataFrame): 每个分箱的详细计算过程
    """
    # 1. 合并数据，确定分箱边界
    all_data = pd.concat([expected, actual], axis=0)
    
    if bucket_type == 'quantile':
        # 等频分箱：基于预期分布的分位数
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(expected, percentiles)
        # 确保边界唯一，避免重复边界导致空箱
        bin_edges = np.unique(bin_edges)
    elif bucket_type == 'width':
        # 等宽分箱
        min_val = min(expected.min(), actual.min())
        max_val = max(expected.max(), actual.max())
        bin_edges = np.linspace(min_val, max_val, bins + 1)
    else:
        raise ValueError("bucket_type 必须为 'quantile' 或 'width'")
    
    # 2. 对预期和实际分布分别分箱
    expected_bins = pd.cut(expected, bins=bin_edges, include_lowest=True, right=False)
    actual_bins = pd.cut(actual, bins=bin_edges, include_lowest=True, right=False)
    
    # 3. 计算每个分箱的样本数
    expected_counts = expected_bins.value_counts().sort_index()
    actual_counts = actual_bins.value_counts().sort_index()
    
    # 4. 计算占比（加微小平滑避免除零）
    epsilon = 1e-6
    expected_pct = (expected_counts + epsilon) / (expected.shape[0] + epsilon * bins)
    actual_pct = (actual_counts + epsilon) / (actual.shape[0] + epsilon * bins)
    
    # 5. 计算每个分箱的PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    
    # 6. 汇总PSI
    psi_value = np.sum(psi_components)
    
    # 7. 构建结果DataFrame
    psi_df = pd.DataFrame({
        '分箱区间': expected_counts.index.astype(str),
        '预期样本数': expected_counts.values,
        '实际样本数': actual_counts.values,
        '预期占比': expected_pct.values,
        '实际占比': actual_pct.values,
        'PSI贡献': psi_components.values
    })
    
    return psi_value, psi_df

# 示例用法
if __name__ == "__main__":
    # 模拟数据：训练集分数（正态分布） vs 当前监控期分数（均值偏移）
    np.random.seed(42)
    train_scores = np.random.normal(loc=600, scale=100, size=10000)
    current_scores = np.random.normal(loc=620, scale=110, size=10000)
    
    expected_series = pd.Series(train_scores, name='预期分数')
    actual_series = pd.Series(current_scores, name='实际分数')
    
    psi, detail_df = calculate_psi(expected_series, actual_series, bins=10)
    
    print(f"整体PSI值: {psi:.4f}")
    print("\n各分箱详细计算：")
    print(detail_df.head(10))

4.3 pandas/numpy关键操作解析

4.4 结果解读与阈值判断

计算得到的PSI值需要结合业务场景进行解读。以下是金融风控领域常用的参考阈值：

4.5 PSI计算流程可视化

下图展示了从原始分数到最终PSI值的完整计算链路，帮助你理解每一步的数据流向。

4.6 分箱方式对比与选择

不同的分箱方式会影响PSI的敏感度和稳定性。以下是两种主流方式的对比：

4.7 批量监控与自动化脚本

在实际生产环境中，通常需要按月或按周批量计算多个特征或模型的PSI。以下是一个批量监控的代码片段：

def batch_psi_monitor(expected_dict, actual_dict, bins=10):
    """
    批量计算多个特征的PSI
    
    参数:
        expected_dict: dict, key为特征名, value为预期分布Series
        actual_dict: dict, key为特征名, value为实际分布Series
    
    返回:
        psi_results: pd.DataFrame, 包含每个特征的PSI值
    """
    results = []
    for feature_name in expected_dict.keys():
        if feature_name not in actual_dict:
            continue
        psi_val, _ = calculate_psi(
            expected_dict[feature_name], 
            actual_dict[feature_name], 
            bins=bins
        )
        results.append({
            '特征名称': feature_name,
            'PSI值': round(psi_val, 4),
            '稳定性判断': '良好' if psi_val < 0.1 else ('轻度偏移' if psi_val < 0.25 else '严重偏移')
        })
    
    return pd.DataFrame(results)

# 示例：批量监控3个特征
expected_features = {
    'score': pd.Series(np.random.normal(600, 100, 10000)),
    'age': pd.Series(np.random.normal(35, 10, 10000)),
    'income': pd.Series(np.random.lognormal(10, 0.5, 10000))
}
actual_features = {
    'score': pd.Series(np.random.normal(620, 110, 10000)),
    'age': pd.Series(np.random.normal(36, 11, 10000)),
    'income': pd.Series(np.random.lognormal(10.2, 0.6, 10000))
}

psi_summary = batch_psi_monitor(expected_features, actual_features)
print(psi_summary)

4.8 PSI异常时的排查路径

当监控到PSI超过阈值时，建议按以下路径进行排查：

1. 检查分箱明细：查看每个分箱的PSI贡献，定位偏移最大的分箱区间。
2. 对比分布图：绘制预期分布与实际分布的直方图或密度图，直观观察偏移方向。
3. 特征层面下钻：如果模型分数PSI异常，进一步计算每个入模特征的PSI，找出导致分数偏移的关键特征。
4. 业务归因：结合催收策略调整、客群结构变化、外部经济环境等因素，判断偏移是否可解释。