3. PSI计算实战：使用Python计算PSI值，处理分箱与边界情况

群体稳定性指标（Population Stability Index, PSI）是反洗钱模型监控中最核心的量化指标。本章将深入PSI的计算原理，并通过Python代码实战演示如何正确处理分箱与边界情况，确保告警规则的准确性。

3.1 PSI计算公式与分箱逻辑

PSI衡量的是两个分布（通常为基准分布与当前分布）之间的差异程度。其核心计算依赖于分箱（Binning）操作，将连续变量或离散变量划分为若干区间，然后比较每个区间内样本占比的变化。

PSI计算公式：

PSI = Σ ( (Actual_i - Expected_i) * ln(Actual_i / Expected_i) )

其中：

• Expected_i：基准分布中第i个分箱的样本占比（如模型训练集或上个月的数据）。
• Actual_i：当前分布中第i个分箱的样本占比（如本月实时数据）。
• ln：自然对数。

3.2 分箱策略与边界处理

分箱方式直接影响PSI计算结果。在反洗钱场景中，常见的分箱策略包括：

边界情况处理原则：

• 空箱处理：当某个分箱在基准或当前分布中样本占比为0时，直接计算ln(0)会导致无穷大。通常采用平滑处理，如给每个箱的计数加一个极小值（如0.5或1e-6）。
• 极值处理：对于超出历史范围的新数据，应将其归入最近的有效分箱或单独设立“异常箱”。
• 分箱数量：建议控制在5~20箱之间。箱数过少会掩盖局部漂移，箱数过多则容易受噪声干扰。

3.3 Python代码实战：PSI计算函数

下面提供一个完整的PSI计算函数，包含分箱、平滑处理和边界保护逻辑。该函数可直接用于反洗钱模型监控流水线。

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, bin_type='quantile', epsilon=1e-6):
    """
    计算群体稳定性指标（PSI）
    
    参数:
        expected: array-like, 基准分布数据
        actual: array-like, 当前分布数据
        bins: int 或 list, 分箱数量或自定义分箱边界
        bin_type: str, 'quantile'(等频) 或 'uniform'(等距)
        epsilon: float, 平滑因子，防止除零或取对数错误
    
    返回:
        psi_value: float, PSI值
        psi_df: DataFrame, 每个分箱的详细计算过程
    """
    # 合并数据确定分箱边界
    all_data = np.concatenate([expected, actual])
    
    if isinstance(bins, int):
        if bin_type == 'quantile':
            # 等频分箱：基于基准分布的分位数
            percentiles = np.linspace(0, 100, bins + 1)[1:-1]
            bin_edges = np.percentile(expected, percentiles)
            bin_edges = np.concatenate([[-np.inf], bin_edges, [np.inf]])
        elif bin_type == 'uniform':
            # 等距分箱
            min_val, max_val = np.min(all_data), np.max(all_data)
            bin_edges = np.linspace(min_val, max_val, bins + 1)
            bin_edges[0] = -np.inf
            bin_edges[-1] = np.inf
        else:
            raise ValueError("bin_type must be 'quantile' or 'uniform'")
    else:
        # 自定义分箱边界
        bin_edges = np.sort(bins)
        bin_edges = np.concatenate([[-np.inf], bin_edges, [np.inf]])
    
    # 计算每个分箱的样本计数
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 计算占比并应用平滑
    expected_pct = (expected_counts + epsilon) / (np.sum(expected_counts) + epsilon * len(expected_counts))
    actual_pct = (actual_counts + epsilon) / (np.sum(actual_counts) + epsilon * len(actual_counts))
    
    # 计算每个分箱的PSI贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi_value = np.sum(psi_components)
    
    # 构建结果DataFrame
    psi_df = pd.DataFrame({
        'bin_start': bin_edges[:-1],
        'bin_end': bin_edges[1:],
        'expected_count': expected_counts,
        'actual_count': actual_counts,
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'psi_component': psi_components
    })
    
    return psi_value, psi_df

# 示例用法
if __name__ == "__main__":
    # 模拟基准分布（如训练集评分）
    np.random.seed(42)
    expected_scores = np.random.normal(loc=0.5, scale=0.15, size=10000)
    # 模拟当前分布（如本月评分，发生轻微漂移）
    actual_scores = np.random.normal(loc=0.55, scale=0.18, size=10000)
    
    psi_val, psi_detail = calculate_psi(expected_scores, actual_scores, bins=10, bin_type='quantile')
    print(f"PSI值: {psi_val:.4f}")
    print("\n分箱详情（前5行）：")
    print(psi_detail.head())

3.4 PSI计算流程可视化

下图展示了PSI计算从数据输入到告警决策的完整流程，重点突出分箱与边界处理环节。

3.5 边界情况专项处理示例

在实际反洗钱数据中，边界情况频繁出现。以下代码演示了如何处理两种典型场景：

# 场景1：空箱处理（基准分布中某个箱占比为0）
expected_data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
actual_data = np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 100])  # 出现极端值

# 使用等距分箱，第5箱在基准中可能为空
psi_val, psi_detail = calculate_psi(expected_data, actual_data, bins=5, bin_type='uniform')
print("场景1 - 含空箱的PSI计算：")
print(f"PSI值: {psi_val:.4f}")
print(psi_detail[['bin_start', 'bin_end', 'expected_count', 'actual_count', 'psi_component']])

# 场景2：自定义分箱处理离散变量（如交易类型编码）
transaction_types = [1, 1, 2, 2, 3, 3, 4, 4, 5, 5]  # 基准
current_types = [1, 1, 1, 2, 2, 3, 3, 4, 6, 6]      # 当前出现新类型6

custom_bins = [1, 2, 3, 4, 5]  # 定义已知类型边界
psi_val2, psi_detail2 = calculate_psi(transaction_types, current_types, bins=custom_bins, bin_type='uniform')
print("\n场景2 - 离散变量含新类别的PSI计算：")
print(f"PSI值: {psi_val2:.4f}")
print(psi_detail2[['bin_start', 'bin_end', 'expected_count', 'actual_count', 'psi_component']])

3.6 分箱策略对比实验

为了直观展示不同分箱策略对PSI结果的影响，我们使用同一组模拟数据（基准分布N(0.5,0.15)，当前分布N(0.55,0.18)）进行对比实验：

3.7 生产环境中的PSI计算架构

下图展示了PSI计算在反洗钱模型监控系统中的架构位置，以及数据流经各组件的关系。

3.8 本章小结

• PSI计算核心：分箱是PSI计算的基础，分箱策略直接影响漂移检测的灵敏度与稳定性。
• 边界处理：空箱平滑（加epsilon）和极值归箱是保证PSI计算鲁棒性的关键步骤。
• 策略选择：等频分箱在反洗钱场景中表现优于等距分箱，推荐箱数10~15。
• 生产实践：PSI计算应作为模型监控流水线的标准化组件，与告警引擎和监控看板集成。