4. PSI计算实战（Python实现）：使用pandas进行分箱、计算实际分布与期望分布、PSI值计算函数封装

本节将使用Python的pandas库，从原始评分数据出发，完整实现PSI计算的三个核心步骤：分箱、分布计算、PSI值计算，并最终封装为可复用的函数。你将掌握如何将理论公式转化为生产级代码。

核心流程：期望分布（训练集）→ 实际分布（验证/监控集）→ 分箱对齐 → 计算各箱占比 → 代入PSI公式 → 输出漂移分数。

4.1 数据准备与分箱策略

假设我们已有两份评分数据：df_train（模型开发时的训练集评分）和 df_monitor（当前监控期的评分）。PSI计算的第一步是对评分进行等频分箱，确保每个箱内的样本数量大致相等。

import pandas as pd
import numpy as np

# 模拟数据
np.random.seed(42)
df_train = pd.DataFrame({'score': np.random.normal(600, 100, 10000)})
df_monitor = pd.DataFrame({'score': np.random.normal(620, 110, 8000)})

# 等频分箱函数：基于训练集确定分箱边界
def equal_freq_bins(series, n_bins=10):
    """返回n_bins个等频分箱的边界列表（含首尾）"""
    percentiles = np.linspace(0, 100, n_bins + 1)
    bins = np.percentile(series, percentiles)
    # 确保边界唯一
    bins = np.unique(bins)
    return bins

# 基于训练集计算分箱边界
bins = equal_freq_bins(df_train['score'], n_bins=10)
print("分箱边界:", bins)
分箱注意事项：
• 分箱边界必须基于期望分布（训练集）确定，监控集使用相同边界。
• 若监控集评分超出边界范围，需将超出部分归入最近箱（左开右闭或左闭右开需统一）。
• 建议箱数在 10~20 之间，箱数过少会掩盖局部漂移，过多则导致统计不稳定。

4.2 计算实际分布与期望分布

使用pandas的cut()函数将评分映射到分箱，然后统计每个箱内的样本占比。

def calc_distribution(df, score_col, bins):
    """计算评分在各分箱中的占比分布"""
    df = df.copy()
    df['bin'] = pd.cut(df[score_col], bins=bins, right=False, include_lowest=True)
    dist = df['bin'].value_counts(normalize=True).sort_index()
    return dist

# 计算期望分布（训练集）
expected_dist = calc_distribution(df_train, 'score', bins)
# 计算实际分布（监控集）
actual_dist = calc_distribution(df_monitor, 'score', bins)

# 合并为DataFrame便于观察
dist_df = pd.DataFrame({
    '期望占比': expected_dist,
    '实际占比': actual_dist
}).fillna(0)  # 若监控集某箱无样本，占比为0
print(dist_df.head())
分箱区间 期望占比（训练集） 实际占比（监控集） 差异
[342.5, 525.0) 0.100 0.085 -0.015
[525.0, 575.0) 0.100 0.092 -0.008
[575.0, 615.0) 0.100 0.103 +0.003
[615.0, 650.0) 0.100 0.115 +0.015
[650.0, 700.0) 0.100 0.108 +0.008
零值处理：当某箱实际占比为0时，直接代入PSI公式会导致除以零错误。建议对占比进行平滑处理（如加极小值 1e-6），或使用 np.log(actual / expected) 前检查分母。

4.3 PSI值计算函数封装

将上述步骤整合为一个完整的PSI计算函数，支持自定义箱数、平滑参数和边界处理。

def calculate_psi(expected_series, actual_series, n_bins=10, eps=1e-6):
    """
    计算PSI（群体稳定性指标）
    
    参数:
        expected_series: 期望分布数据（训练集评分）
        actual_series:   实际分布数据（监控集评分）
        n_bins:          分箱数量，默认10
        eps:             平滑因子，防止除零
        
    返回:
        psi_value: PSI值
        psi_df:     包含各箱详细信息的DataFrame
    """
    # 1. 基于期望数据确定分箱边界
    bins = np.percentile(expected_series, np.linspace(0, 100, n_bins + 1))
    bins = np.unique(bins)
    
    # 2. 计算期望分布
    expected_binned = pd.cut(expected_series, bins=bins, right=False, include_lowest=True)
    expected_dist = expected_binned.value_counts(normalize=True).sort_index()
    
    # 3. 计算实际分布（使用相同边界）
    actual_binned = pd.cut(actual_series, bins=bins, right=False, include_lowest=True)
    actual_dist = actual_binned.value_counts(normalize=True).sort_index()
    
    # 4. 合并并填充缺失箱
    psi_df = pd.DataFrame({
        'expected_pct': expected_dist,
        'actual_pct': actual_dist
    }).fillna(0)
    
    # 5. 平滑处理
    psi_df['expected_pct'] = psi_df['expected_pct'].clip(lower=eps)
    psi_df['actual_pct'] = psi_df['actual_pct'].clip(lower=eps)
    
    # 6. 计算各箱PSI分量
    psi_df['psi_component'] = (psi_df['actual_pct'] - psi_df['expected_pct']) * \
                              np.log(psi_df['actual_pct'] / psi_df['expected_pct'])
    
    # 7. 汇总PSI值
    psi_value = psi_df['psi_component'].sum()
    
    return psi_value, psi_df

# 使用示例
psi_value, psi_detail = calculate_psi(df_train['score'], df_monitor['score'], n_bins=10)
print(f"PSI值: {psi_value:.4f}")
print(psi_detail[['expected_pct', 'actual_pct', 'psi_component']].head())
输出解读：
• PSI < 0.1：无显著漂移，模型稳定。
• 0.1 ≤ PSI < 0.25：轻度漂移，需关注并排查原因。
• PSI ≥ 0.25：严重漂移，建议立即进行模型重构或校准。

4.4 PSI计算流程可视化

下图展示了从原始评分到PSI值的完整计算流程，帮助你理解各步骤之间的数据流转关系。

PSI计算流程（Python实现） 训练集评分 监控集评分 等频分箱（基于训练集） np.percentile → 边界列表 分布计算 pd.cut → value_counts(normalize=True) PSI公式计算 Σ (实际-期望) * ln(实际/期望) PSI值 + 分箱明细 分箱边界仅由训练集确定 监控集使用相同边界映射

4.5 函数封装要点与扩展

  • 参数化设计：n_bins、eps 作为函数参数，便于不同场景调优。
  • 边界一致性：训练集和监控集必须使用完全相同的 bins 数组，否则PSI值失真。
  • 批量计算：可扩展为接收多个特征列，循环调用 calculate_psi() 生成特征级PSI报告。
  • 可视化集成：返回的 psi_df 可直接用于绘制堆叠柱状图或PSI分量图。
生产环境建议：
1. 将PSI计算函数封装为独立的Python模块（如 psi_monitor.py），便于定时任务调用。
2. 结合日志模块记录每次计算的PSI值、时间戳和分箱明细，形成历史趋势库。
3. 当PSI超过阈值时，自动触发告警邮件或企业微信通知。

4.6 完整代码示例（单文件）

import pandas as pd
import numpy as np

def calculate_psi(expected_series, actual_series, n_bins=10, eps=1e-6):
    """完整PSI计算函数"""
    bins = np.percentile(expected_series, np.linspace(0, 100, n_bins + 1))
    bins = np.unique(bins)
    
    expected_binned = pd.cut(expected_series, bins=bins, right=False, include_lowest=True)
    expected_dist = expected_binned.value_counts(normalize=True).sort_index()
    
    actual_binned = pd.cut(actual_series, bins=bins, right=False, include_lowest=True)
    actual_dist = actual_binned.value_counts(normalize=True).sort_index()
    
    psi_df = pd.DataFrame({'expected_pct': expected_dist, 'actual_pct': actual_dist}).fillna(0)
    psi_df['expected_pct'] = psi_df['expected_pct'].clip(lower=eps)
    psi_df['actual_pct'] = psi_df['actual_pct'].clip(lower=eps)
    psi_df['psi_component'] = (psi_df['actual_pct'] - psi_df['expected_pct']) * \
                              np.log(psi_df['actual_pct'] / psi_df['expected_pct'])
    
    return psi_df['psi_component'].sum(), psi_df

# 快速测试
if __name__ == "__main__":
    train = pd.Series(np.random.normal(600, 100, 10000))
    monitor = pd.Series(np.random.normal(620, 110, 8000))
    psi, detail = calculate_psi(train, monitor)
    print(f"PSI = {psi:.4f}")
← 上一章 📖 返回目录 下一章 →