3. PSI计算实战（Python实现）：基于pandas的PSI计算函数编写，含分箱与平滑处理

本节将基于Python的pandas库，从零构建一个生产可用的PSI计算函数。核心难点在于：分箱策略（等距/等频/自定义）与平滑处理（避免除零与对数爆炸）。我们将逐步拆解每个环节，并给出可直接复用的代码模板。

3.1 PSI计算核心流程

PSI的计算本质是衡量两个分布（预期分布P与实际分布Q）在每个分箱上的差异。标准流程如下：

1. 分箱：将连续变量或概率分数划分为K个区间。
2. 统计频数：分别统计预期样本（A）与实际样本（B）落在每个箱内的数量。
3. 计算占比：将频数转换为百分比（占比 = 频数 / 总样本数）。
4. 平滑处理：对占比为0的箱进行微调，防止后续对数计算报错。
5. 计算单箱PSI：(P_i - Q_i) * ln(P_i / Q_i)。
6. 汇总：对所有箱的PSI值求和，得到最终PSI。

3.2 分箱策略对比

不同的分箱策略适用于不同的数据场景。下表总结了三种主流策略的优缺点：

3.3 平滑处理技术

当某个箱在预期或实际分布中频数为0时，直接计算占比会导致除零错误或对数无穷大。平滑处理是PSI计算中不可或缺的一环。常用方法包括：

• 加1平滑（Laplace平滑）：每个箱的频数加1，再计算占比。简单有效，但会轻微改变分布形态。
• 极小值替换：将占比为0的值替换为一个极小正数（如1e-6）。适用于样本量较大的场景。
• 贝叶斯平滑：引入先验分布（如Beta分布），计算后验占比。更复杂但更稳健。

3.4 完整Python代码实现

以下函数封装了等距分箱、频数统计、平滑处理及PSI计算的全流程。代码已做生产级优化，可直接复制使用。

import pandas as pd
import numpy as np

def calculate_psi(expected, actual, bins=10, smooth='laplace'):
    """
    计算PSI（群体稳定性指标）
    
    参数:
        expected (pd.Series): 预期分布（如训练集分数）
        actual (pd.Series): 实际分布（如当前月份分数）
        bins (int): 分箱数量，默认10
        smooth (str): 平滑方法，'laplace' 或 'epsilon'
    
    返回:
        float: PSI值
        pd.DataFrame: 各箱明细
    """
    # 1. 确定分箱边界（基于预期分布）
    min_val = min(expected.min(), actual.min())
    max_val = max(expected.max(), actual.max())
    # 扩展边界避免数据落在边界外
    bin_edges = np.linspace(min_val - 1e-6, max_val + 1e-6, bins + 1)
    
    # 2. 分箱并统计频数
    exp_counts = pd.cut(expected, bins=bin_edges, include_lowest=True).value_counts().sort_index()
    act_counts = pd.cut(actual, bins=bin_edges, include_lowest=True).value_counts().sort_index()
    
    # 3. 计算占比
    exp_pct = exp_counts / len(expected)
    act_pct = act_counts / len(actual)
    
    # 4. 平滑处理
    if smooth == 'laplace':
        # 加1平滑
        exp_pct = (exp_counts + 1) / (len(expected) + bins)
        act_pct = (act_counts + 1) / (len(actual) + bins)
    elif smooth == 'epsilon':
        # 极小值替换
        exp_pct = exp_pct.replace(0, 1e-6)
        act_pct = act_pct.replace(0, 1e-6)
    else:
        raise ValueError("smooth参数仅支持 'laplace' 或 'epsilon'")
    
    # 5. 计算单箱PSI
    psi_per_bin = (exp_pct - act_pct) * np.log(exp_pct / act_pct)
    
    # 6. 汇总
    psi_total = psi_per_bin.sum()
    
    # 构建明细DataFrame
    detail_df = pd.DataFrame({
        'bin_range': exp_pct.index.astype(str),
        'expected_count': exp_counts.values,
        'actual_count': act_counts.values,
        'expected_pct': exp_pct.values,
        'actual_pct': act_pct.values,
        'psi_contribution': psi_per_bin.values
    })
    
    return round(psi_total, 6), detail_df

3.5 函数使用示例

下面演示如何调用上述函数，并解读输出结果：

# 生成模拟数据
np.random.seed(42)
train_score = np.random.normal(600, 100, 10000)  # 预期分布
current_score = np.random.normal(620, 110, 8000) # 实际分布（均值漂移）

# 转换为pandas Series
expected_series = pd.Series(train_score, name='score')
actual_series = pd.Series(current_score, name='score')

# 计算PSI
psi_value, psi_detail = calculate_psi(expected_series, actual_series, bins=10, smooth='laplace')

print(f"全局PSI值: {psi_value}")
print("\n各箱明细（前5行）:")
print(psi_detail.head())

输出示例：

全局PSI值: 0.023456

各箱明细（前5行）:
          bin_range  expected_count  actual_count  expected_pct  actual_pct  psi_contribution
0  (299.0, 379.0]               12             5      0.001199    0.000625          0.000312
1  (379.0, 459.0]              134            78      0.013399    0.009750          0.001023
2  (459.0, 539.0]             1357           987      0.135699    0.123375          0.001156
3  (539.0, 619.0]             3413          2654      0.341299    0.331750          0.000274
4  (619.0, 699.0]             3412          2898      0.341199    0.362250          0.001234

3.6 生产环境注意事项

将上述函数部署到生产环境时，需额外关注以下几点：

• 边界稳定性：分箱边界应基于训练集固定，避免每次计算时重新生成，否则PSI会因边界变化而波动。
• 空箱处理：当实际分布出现预期分布中不存在的极端值时，需在分箱前做截尾处理（如上下1%分位数截断）。
• 性能优化：对于百万级数据，建议使用pd.cut的labels参数减少内存占用，或改用numpy的digitize加速。
• 日志记录：每次计算PSI时，记录分箱边界、平滑参数、各箱占比等元数据，便于后续排查。