3. PSI计算实战：使用Python实现PSI计算函数，处理缺失值与异常值

在完成PSI的理论推导与阈值设定后，本节将进入工程落地环节。我们将使用Python实现一个健壮的PSI计算函数，重点解决实际数据中常见的缺失值、零分母和极端分布问题。一个未经异常处理的PSI函数在生产环境中极易产生NaN或无穷大，导致误告警或漏告警。

3.1 标准PSI计算函数（基础版）

以下函数实现了PSI的核心公式：PSI = Σ (实际占比 - 预期占比) * ln(实际占比 / 预期占比)。该版本假设输入数据已经完成分箱且无缺失值。

import numpy as np
import pandas as pd

def psi_base(expected, actual, bins=10):
    """
    基础PSI计算函数（无异常处理）
    
    Parameters:
    expected: array-like, 预期分布（训练集）
    actual: array-like, 实际分布（生产集）
    bins: int, 分箱数量
    
    Returns:
    float, PSI值
    """
    # 等宽分箱
    _, bin_edges = np.histogram(expected, bins=bins)
    
    # 计算各箱占比
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    expected_pct = expected_counts / len(expected)
    actual_pct = actual_counts / len(actual)
    
    # 计算PSI
    psi_value = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return psi_value
⚠ 生产风险：上述函数在以下场景会直接崩溃或返回NaN：
1. 某分箱内预期占比为0 → 分母为零，ln(0)无定义。
2. 某分箱内实际占比为0 → ln(0)无定义。
3. 输入数据包含NaN → 分箱结果异常。
4. 所有数据落入同一分箱 → 其他分箱占比为0。

3.2 健壮版PSI函数（含缺失值与异常值处理）

生产级PSI函数需要处理以下三类问题：

问题类型 影响 处理策略
缺失值（NaN） 分箱计数错误，PSI计算失真 单独成箱或剔除后记录缺失率变化
零占比分箱 ln(0)导致NaN或无穷大 添加极小平滑值（如1e-6）
极端离群值 分箱边界被拉偏，PSI虚高 基于百分位数截断（如1%-99%） 
def psi_robust(expected, actual, bins=10, smooth=1e-6, 
               clip_range=(0.01, 0.99), handle_missing='separate'):
    """
    健壮PSI计算函数（生产级）
    
    Parameters:
    expected: array-like, 预期分布
    actual: array-like, 实际分布
    bins: int, 分箱数量
    smooth: float, 平滑值，防止零占比
    clip_range: tuple, 截断百分位范围
    handle_missing: str, 'separate' | 'drop' | 'raise'
    
    Returns:
    dict: {'psi': float, 'missing_rate_expected': float, 
           'missing_rate_actual': float, 'bins_used': int}
    """
    # 1. 缺失值处理
    exp_missing = np.isnan(expected).sum() / len(expected)
    act_missing = np.isnan(actual).sum() / len(actual)
    
    if handle_missing == 'raise' and (exp_missing > 0 or act_missing > 0):
        raise ValueError(f"数据包含缺失值: 预期{exp_missing:.2%}, 实际{act_missing:.2%}")
    
    if handle_missing == 'drop':
        expected = expected[~np.isnan(expected)]
        actual = actual[~np.isnan(actual)]
    elif handle_missing == 'separate':
        # 将缺失值作为独立分箱处理
        exp_non_missing = expected[~np.isnan(expected)]
        act_non_missing = actual[~np.isnan(actual)]
    else:
        exp_non_missing = expected
        act_non_missing = actual
    
    # 2. 极端值截断（基于百分位数）
    lower = np.percentile(exp_non_missing, clip_range[0] * 100)
    upper = np.percentile(exp_non_missing, clip_range[1] * 100)
    
    exp_clipped = np.clip(exp_non_missing, lower, upper)
    act_clipped = np.clip(act_non_missing, lower, upper)
    
    # 3. 分箱（基于预期分布的边界）
    _, bin_edges = np.histogram(exp_clipped, bins=bins)
    
    exp_counts, _ = np.histogram(exp_clipped, bins=bin_edges)
    act_counts, _ = np.histogram(act_clipped, bins=bin_edges)
    
    # 4. 计算占比 + 平滑处理
    exp_pct = (exp_counts + smooth) / (len(exp_clipped) + smooth * bins)
    act_pct = (act_counts + smooth) / (len(act_clipped) + smooth * bins)
    
    # 5. 计算PSI
    psi_value = np.sum((act_pct - exp_pct) * np.log(act_pct / exp_pct))
    
    # 6. 如果采用separate模式，加入缺失率漂移惩罚项
    if handle_missing == 'separate':
        missing_psi = (act_missing - exp_missing) * np.log(
            (act_missing + smooth) / (exp_missing + smooth)
        )
        psi_value += missing_psi
    
    return {
        'psi': round(psi_value, 6),
        'missing_rate_expected': round(exp_missing, 4),
        'missing_rate_actual': round(act_missing, 4),
        'bins_used': bins
    }
🔧 平滑值选择建议：
  • 1e-6：适用于样本量 > 10万的大规模场景，对PSI影响可忽略。
  • 1e-4：适用于样本量 1万~10万的中等规模场景。
  • 1e-3：适用于样本量 < 1万的小样本场景，防止过度平滑导致PSI被压低。

3.3 PSI计算流程与异常处理决策树

下图展示了PSI计算过程中各异常处理分支的决策逻辑，帮助你在实际工程中快速定位问题环节。

PSI计算异常处理决策流程 输入 expected / actual 是否存在缺失值？ 是 缺失值处理 separate / drop / raise 否 极端值截断 (1%-99%) 等宽分箱 (bins=10) 存在零占比分箱？ 是 添加平滑值 1e-6 否 输出 PSI 值

3.4 实战案例：信贷评分卡PSI监控

以下案例模拟了信贷评分卡模型在生产环境中的PSI计算，包含缺失值注入和极端值场景。

# 生成模拟数据
np.random.seed(42)
n_train = 50000
n_prod = 50000

# 训练集：正态分布评分
train_scores = np.random.normal(loc=600, scale=100, size=n_train)
train_scores = np.clip(train_scores, 300, 900)

# 生产集：分布偏移 + 缺失值 + 极端值
prod_scores = np.random.normal(loc=580, scale=120, size=n_prod)
prod_scores = np.clip(prod_scores, 300, 900)

# 注入5%缺失值
missing_idx = np.random.choice(n_prod, size=int(n_prod * 0.05), replace=False)
prod_scores[missing_idx] = np.nan

# 注入1%极端值
extreme_idx = np.random.choice(n_prod, size=int(n_prod * 0.01), replace=False)
prod_scores[extreme_idx] = np.random.uniform(0, 300, size=len(extreme_idx))

# 计算PSI
result = psi_robust(
    expected=train_scores, 
    actual=prod_scores, 
    bins=10, 
    smooth=1e-6,
    clip_range=(0.01, 0.99),
    handle_missing='separate'
)

print(f"PSI值: {result['psi']}")
print(f"预期缺失率: {result['missing_rate_expected']:.4f}")
print(f"实际缺失率: {result['missing_rate_actual']:.4f}")
print(f"使用分箱数: {result['bins_used']}")
💡 输出解读：
若PSI > 0.25，说明生产分布发生显著漂移，需要排查特征工程或数据采集环节。
若缺失率从0%上升到5%以上，即使PSI未超标，也应触发数据质量告警。

3.5 常见踩坑与避坑指南

  • 坑1：分箱边界不一致——生产集和训练集使用各自的分箱边界，导致PSI虚高。应始终使用训练集的分箱边界固定生产集的计算。
  • 坑2：平滑值过大——当样本量较小时，过大的平滑值会压低真实漂移信号，建议根据样本量动态调整。
  • 坑3：忽略缺失率变化——缺失率从1%上升到10%本身就是重大漂移信号，应在PSI之外单独监控缺失率。
  • 坑4：对类别特征直接使用等宽分箱——类别特征应使用频次分箱或WOE分箱，等宽分箱会导致分箱边界无意义。
📊 生产环境最佳实践：
  1. 将PSI计算封装为微服务，支持批量与实时两种调用模式。
  2. 每次计算时记录分箱边界、缺失率、平滑值等元数据，便于回溯。
  3. 设置双重告警：PSI > 0.25 触发模型漂移告警；缺失率变化 > 3% 触发数据质量告警。
  4. 定期（如每周）重新计算训练集的分箱边界，适应特征分布的长期缓慢变化。
← 上一章 📖 返回目录 下一章 →