4. PSI计算Python实现（上）：使用pandas进行基础PSI计算，单特征PSI函数编写，处理缺失值与异常值

在掌握了PSI的理论公式与分箱逻辑后，本章进入实战环节。我们将使用Python的pandas库，从零开始构建一个稳定、可复用的单特征PSI计算函数。本节重点解决三个核心问题：基础PSI向量化计算、缺失值与异常值的处理策略，以及函数封装的最佳实践。

4.1 基础PSI计算：从公式到pandas向量化

PSI的核心公式为：

PSI = Σ ( (P_i - Q_i) * ln(P_i / Q_i) )

其中P_i为实际分布（当前样本）在第i个分箱的比例，Q_i为预期分布（建模样本）在第i个分箱的比例。使用pandas实现时，应避免逐行循环，直接利用向量化运算。

基础计算步骤：

1. 将连续特征分箱（等距、等频或自定义分箱）。
2. 统计每个分箱在预期分布和实际分布中的样本计数。
3. 计算每个分箱的比例（计数/总样本数）。
4. 对比例为零的分箱做平滑处理（避免除零或ln(0)）。
5. 代入公式计算PSI值。

以下是一个使用pandas实现的基础PSI计算代码：

import pandas as pd
import numpy as np

def calculate_psi_base(expected, actual, bins=10):
    """
    基础PSI计算函数（等频分箱）
    
    Parameters:
    expected: array-like, 预期分布（建模样本）
    actual: array-like, 实际分布（当前样本）
    bins: int, 分箱数量
    
    Returns:
    float, PSI值
    """
    # 使用预期分布的分位数确定分箱边界
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected, percentiles)
    # 处理边界重复的情况（当数据中有大量重复值时）
    bin_edges = np.unique(bin_edges)
    
    # 对预期和实际分布进行分箱
    expected_binned = pd.cut(expected, bins=bin_edges, include_lowest=True)
    actual_binned = pd.cut(actual, bins=bin_edges, include_lowest=True)
    
    # 统计每个分箱的计数
    expected_counts = expected_binned.value_counts().sort_index()
    actual_counts = actual_binned.value_counts().sort_index()
    
    # 计算比例，并做平滑处理（加一个小常数避免除零）
    epsilon = 1e-6
    expected_pct = expected_counts / expected_counts.sum() + epsilon
    actual_pct = actual_counts / actual_counts.sum() + epsilon
    
    # 计算PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return psi

4.2 单特征PSI函数编写：工程化封装

基础函数可以工作，但在生产环境中需要更强的鲁棒性。一个工程化的PSI函数应包含以下特性：

工程化PSI函数实现：

def calculate_psi_robust(expected, actual, bins=10, bins_type='quantile',
                         epsilon=1e-6, handle_missing=True, return_details=False):
    """
    工程化PSI计算函数，支持缺失值处理、异常值处理和多种分箱方式。
    
    Parameters:
    expected: array-like, 预期分布
    actual: array-like, 实际分布
    bins: int or list, 分箱数量或自定义边界
    bins_type: str, 'quantile'(等频) 或 'uniform'(等距)
    epsilon: float, 平滑常数
    handle_missing: bool, 是否单独处理缺失值
    return_details: bool, 是否返回分箱明细
    
    Returns:
    float or dict, PSI值或包含明细的字典
    """
    # 转换为pandas Series
    exp_series = pd.Series(expected).dropna() if handle_missing else pd.Series(expected)
    act_series = pd.Series(actual).dropna() if handle_missing else pd.Series(actual)
    
    # 处理缺失值占比
    missing_info = {}
    if handle_missing:
        exp_missing = pd.Series(expected).isna().mean()
        act_missing = pd.Series(actual).isna().mean()
        missing_info = {'exp_missing_pct': exp_missing, 'act_missing_pct': act_missing}
    
    # 确定分箱边界
    if isinstance(bins, int):
        if bins_type == 'quantile':
            percentiles = np.linspace(0, 100, bins + 1)
            bin_edges = np.percentile(exp_series, percentiles)
        elif bins_type == 'uniform':
            bin_edges = np.linspace(exp_series.min(), exp_series.max(), bins + 1)
        else:
            raise ValueError("bins_type must be 'quantile' or 'uniform'")
        bin_edges = np.unique(bin_edges)
    else:
        bin_edges = np.array(bins)
    
    # 处理异常值：将超出边界的值clip到边界内
    exp_clipped = np.clip(exp_series, bin_edges[0], bin_edges[-1])
    act_clipped = np.clip(act_series, bin_edges[0], bin_edges[-1])
    
    # 分箱
    exp_binned = pd.cut(exp_clipped, bins=bin_edges, include_lowest=True)
    act_binned = pd.cut(act_clipped, bins=bin_edges, include_lowest=True)
    
    # 统计计数
    exp_counts = exp_binned.value_counts().sort_index()
    act_counts = act_binned.value_counts().sort_index()
    
    # 计算比例 + 平滑
    exp_pct = exp_counts / exp_counts.sum() + epsilon
    act_pct = act_counts / act_counts.sum() + epsilon
    
    # 归一化（确保平滑后比例之和为1）
    exp_pct = exp_pct / exp_pct.sum()
    act_pct = act_pct / act_pct.sum()
    
    # 计算PSI
    psi_value = np.sum((act_pct - exp_pct) * np.log(act_pct / exp_pct))
    
    if return_details:
        details = {
            'psi': psi_value,
            'bin_edges': bin_edges,
            'exp_pct': exp_pct,
            'act_pct': act_pct,
            'exp_counts': exp_counts,
            'act_counts': act_counts,
            'missing_info': missing_info
        }
        return details
    return psi_value

4.3 缺失值与异常值处理策略

缺失值和异常值是PSI计算中常见的陷阱，处理不当会导致PSI值失真。以下是推荐的策略：

缺失值处理

• 策略一：单独分箱 — 将缺失值作为一个独立的分箱，计算其占比变化。适用于缺失值有业务含义的场景。
• 策略二：剔除后计算 — 在计算PSI前剔除缺失值，但需额外监控缺失率变化。适用于缺失值随机且占比小的场景。
• 策略三：填充后计算 — 使用均值/中位数/众数填充缺失值。适用于缺失值较少且填充方式合理的场景。

异常值处理

• 策略一：截断（Clipping） — 将超出分箱边界的值强制设为边界值。适用于异常值极少且边界合理的场景。
• 策略二：单独分箱 — 将异常值归入最左或最右分箱，或单独设立“异常值分箱”。适用于异常值有业务意义的场景。
• 策略三：分位数分箱 — 使用等频分箱，异常值自然被归入首尾分箱。这是最常用的方法。

4.4 实战案例：使用PSI函数监控特征稳定性

以下是一个完整的实战案例，展示如何使用上述函数监控模型特征的PSI：

# 模拟数据
np.random.seed(42)
expected_data = np.random.normal(0, 1, 10000)  # 建模样本
actual_data = np.random.normal(0.2, 1.1, 10000)  # 当前样本（有轻微漂移）

# 引入缺失值
expected_data[np.random.choice(10000, 200, replace=False)] = np.nan
actual_data[np.random.choice(10000, 500, replace=False)] = np.nan

# 计算PSI（返回明细）
result = calculate_psi_robust(
    expected=expected_data,
    actual=actual_data,
    bins=10,
    bins_type='quantile',
    handle_missing=True,
    return_details=True
)

print(f"PSI值: {result['psi']:.4f}")
print(f"预期分布缺失率: {result['missing_info']['exp_missing_pct']:.2%}")
print(f"实际分布缺失率: {result['missing_info']['act_missing_pct']:.2%}")

# 输出各分箱占比
psi_df = pd.DataFrame({
    '分箱区间': result['exp_pct'].index.astype(str),
    '预期占比': result['exp_pct'].values,
    '实际占比': result['act_pct'].values,
    '贡献度': (result['act_pct'] - result['exp_pct']) * np.log(result['act_pct'] / result['exp_pct'])
})
print("\n分箱明细:")
print(psi_df.round(4))

输出示例：

PSI值: 0.0358
预期分布缺失率: 2.00%
实际分布缺失率: 5.00%

分箱明细:
              分箱区间  预期占比  实际占比  贡献度
0  (-2.326, -1.282]  0.1002  0.0892  0.0013
1  (-1.282, -0.674]  0.1000  0.0921  0.0007
2  (-0.674, -0.201]  0.1001  0.0954  0.0002
3  (-0.201, 0.201]   0.0998  0.1012  0.0000
4  (0.201, 0.674]    0.1000  0.1087  0.0007
5  (0.674, 1.282]    0.1002  0.1156  0.0021
6  (1.282, 2.326]    0.0999  0.1089  0.0008
7  (2.326, inf]      0.0998  0.0891  0.0012

4.5 常见陷阱与避坑指南

• 陷阱1：分箱边界不一致 — 实际分布使用了与预期分布不同的分箱边界。解决方案：始终使用预期分布的分位数确定边界。
• 陷阱2：零分箱未处理 — 某个分箱在预期或实际分布中计数为零，导致除零错误。解决方案：加入平滑常数或合并零计数分箱。
• 陷阱3：缺失值被忽略 — 缺失值被pandas自动丢弃，导致样本量减少且未监控缺失率变化。解决方案：显式处理缺失值，并单独监控缺失率。
• 陷阱4：异常值扭曲边界 — 极端异常值导致分箱边界被拉宽，大部分样本集中在少数分箱。解决方案：使用分位数分箱或对异常值进行截断。
• 陷阱5：分箱数量不当 — 分箱过多导致每个分箱样本量不足，分箱过少则灵敏度下降。解决方案：根据样本量调整分箱数，通常建议10-20个分箱。