3. PSI 计算实战：使用 Python 计算 PSI、分箱策略对 PSI 的影响、PSI 计算代码实现

本章进入 PSI 计算的实战环节。你将掌握如何用 Python 从零实现 PSI 计算，并深入理解分箱策略对 PSI 结果的关键影响。通过代码与可视化结合，建立对 PSI 稳定性的量化感知。

3.1 PSI 计算核心公式与分箱逻辑

PSI 的计算基于“分箱-占比-对数-求和”四步。其数学本质是衡量两个分布在各分箱上的偏差程度。

3.2 分箱策略对 PSI 的影响

分箱策略是 PSI 计算中最具主观性的环节，直接影响 PSI 的数值和告警灵敏度。以下是三种主流策略的对比：

3.3 PSI 计算代码实现

以下 Python 代码实现了完整的 PSI 计算函数，支持等距、等频和自定义分箱策略。代码包含边界处理、零值保护和结果输出。

import numpy as np
import pandas as pd

def calculate_psi(actual, expected, bins=10, strategy='equal_width', custom_bins=None):
    """
    计算 PSI (Population Stability Index)
    
    参数:
        actual: array-like, 实际分布数据
        expected: array-like, 预期分布数据
        bins: int, 分箱数量 (默认10)
        strategy: str, 分箱策略 ('equal_width', 'equal_freq', 'custom')
        custom_bins: list, 自定义分箱边界 (仅 strategy='custom' 时使用)
    
    返回:
        psi: float, PSI 值
        bin_df: DataFrame, 各分箱的详细计算过程
    """
    # 合并数据确定分箱边界
    all_data = np.concatenate([actual, expected])
    
    if strategy == 'equal_width':
        # 等距分箱
        bin_edges = np.linspace(np.min(all_data), np.max(all_data), bins + 1)
    elif strategy == 'equal_freq':
        # 等频分箱 (基于预期分布)
        bin_edges = np.percentile(expected, np.linspace(0, 100, bins + 1))
        # 确保边界唯一
        bin_edges = np.unique(bin_edges)
        if len(bin_edges) < 2:
            bin_edges = np.array([np.min(all_data), np.max(all_data)])
    elif strategy == 'custom':
        # 自定义分箱
        if custom_bins is None:
            raise ValueError("自定义分箱需提供 custom_bins 参数")
        bin_edges = np.array(custom_bins)
    else:
        raise ValueError("分箱策略仅支持: equal_width, equal_freq, custom")
    
    # 分箱并计算占比
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    
    actual_pct = actual_counts / len(actual)
    expected_pct = expected_counts / len(expected)
    
    # 零值保护：用极小值替代0，避免 ln(0) 错误
    epsilon = 1e-6
    actual_pct = np.where(actual_pct == 0, epsilon, actual_pct)
    expected_pct = np.where(expected_pct == 0, epsilon, expected_pct)
    
    # 计算每个分箱的 PSI 贡献
    psi_components = (actual_pct - expected_pct) * np.log(actual_pct / expected_pct)
    psi = np.sum(psi_components)
    
    # 构建结果 DataFrame
    bin_labels = [f"箱{i+1}" for i in range(len(bin_edges)-1)]
    bin_df = pd.DataFrame({
        '分箱': bin_labels,
        '区间下界': bin_edges[:-1],
        '区间上界': bin_edges[1:],
        '实际占比': actual_pct,
        '预期占比': expected_pct,
        'PSI贡献': psi_components
    })
    
    return psi, bin_df

# ========== 使用示例 ==========
# 生成模拟数据
np.random.seed(42)
expected_scores = np.random.normal(50, 15, 10000)  # 预期分布
actual_scores = np.random.normal(55, 18, 10000)    # 实际分布 (发生漂移)

# 计算 PSI (等频分箱)
psi_value, psi_detail = calculate_psi(
    actual=actual_scores, 
    expected=expected_scores, 
    bins=10, 
    strategy='equal_freq'
)

print(f"PSI 值: {psi_value:.4f}")
print("\n各分箱详情:")
print(psi_detail.to_string(index=False))

3.4 代码输出解读与告警阈值

运行上述代码后，输出结果包含 PSI 总分和各分箱的贡献明细。反洗钱模型看板中，PSI 的告警阈值通常参考以下标准：

3.5 分箱策略对比实验

下面通过一个对比实验，直观展示不同分箱策略对同一组数据计算出的 PSI 差异：

# 对比三种分箱策略的 PSI 结果
strategies = ['equal_width', 'equal_freq', 'custom']
custom_bins = [0, 20, 40, 60, 80, 100]  # 自定义边界

results = []
for strategy in strategies:
    if strategy == 'custom':
        psi_val, _ = calculate_psi(actual_scores, expected_scores, 
                                   strategy=strategy, custom_bins=custom_bins)
    else:
        psi_val, _ = calculate_psi(actual_scores, expected_scores, 
                                   bins=10, strategy=strategy)
    results.append({'分箱策略': strategy, 'PSI值': round(psi_val, 4)})

result_df = pd.DataFrame(results)
print(result_df.to_string(index=False))