2. PSI计算实战：使用Python计算PSI的完整代码实现

在掌握PSI的理论定义后，本节将进入实战环节。我们将使用Python从零实现PSI计算，涵盖等频分箱、概率分布计算和最终PSI值求解。完整的代码实现将帮助你理解每一步的数学含义，并为后续的阈值调优打下坚实基础。

2.1 PSI计算核心流程

PSI的计算遵循一个标准化的四步流程，每一步都直接影响最终结果的准确性。下图展示了从原始数据到PSI值的完整链路：

2.2 完整代码实现

以下代码实现了PSI计算的完整流程，包含等频分箱、概率分布计算和PSI值求解。代码中加入了详细的注释，便于理解每一步的数学含义。

import numpy as np
import pandas as pd
from scipy import stats

def calculate_psi(expected, actual, bins=10, bucket_type='quantile'):
    """
    计算PSI（群体稳定性指标）
    
    Parameters:
    -----------
    expected : array-like
        预期分布数据（通常为基准期数据）
    actual : array-like
        实际分布数据（通常为监控期数据）
    bins : int, default=10
        分箱数量
    bucket_type : str, default='quantile'
        分箱方式：'quantile' 等频分箱，'equal' 等距分箱
    
    Returns:
    --------
    psi_value : float
        PSI值
    breakdown : pd.DataFrame
        每个分箱的详细计算过程
    """
    
    # 1. 数据预处理：转换为numpy数组并去除NaN
    expected = np.array(expected).flatten()
    actual = np.array(actual).flatten()
    expected = expected[~np.isnan(expected)]
    actual = actual[~np.isnan(actual)]
    
    # 2. 基于预期数据确定分箱边界
    if bucket_type == 'quantile':
        # 等频分箱：每个分箱包含相同数量的样本
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(expected, percentiles)
    elif bucket_type == 'equal':
        # 等距分箱：每个分箱的数值范围相同
        bin_edges = np.linspace(expected.min(), expected.max(), bins + 1)
    else:
        raise ValueError("bucket_type must be 'quantile' or 'equal'")
    
    # 确保分箱边界唯一（处理重复值）
    bin_edges = np.unique(bin_edges)
    
    # 3. 计算每个分箱中的样本数量
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 4. 计算概率分布（添加平滑处理防止除零）
    epsilon = 1e-10
    expected_pct = expected_counts / (expected_counts.sum() + epsilon)
    actual_pct = actual_counts / (actual_counts.sum() + epsilon)
    
    # 5. 计算每个分箱的PSI贡献
    # PSI = (实际比例 - 预期比例) * ln(实际比例 / 预期比例)
    psi_components = []
    for i in range(len(expected_pct)):
        e = expected_pct[i]
        a = actual_pct[i]
        
        # 零值处理：如果预期比例为0，使用极小值替代
        if e == 0:
            e = epsilon
        if a == 0:
            a = epsilon
            
        component = (a - e) * np.log(a / e)
        psi_components.append(component)
    
    # 6. 汇总PSI值
    psi_value = np.sum(psi_components)
    
    # 7. 构建详细结果DataFrame
    breakdown = 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, breakdown

# ============ 使用示例 ============
if __name__ == "__main__":
    # 生成模拟数据
    np.random.seed(42)
    
    # 基准期数据：正态分布，均值0，标准差1
    expected_data = np.random.normal(0, 1, 10000)
    
    # 监控期数据：均值偏移0.2，标准差不变
    actual_data = np.random.normal(0.2, 1, 10000)
    
    # 计算PSI
    psi, breakdown = calculate_psi(expected_data, actual_data, bins=10)
    
    print(f"PSI值: {psi:.6f}")
    print("\n分箱详细计算过程：")
    print(breakdown.to_string(index=False))
    
    # 输出示例（实际运行结果）：
    # PSI值: 0.023847
    # 
    # 分箱详细计算过程：
    #  bin_start   bin_end  expected_count  actual_count  expected_pct  actual_pct  psi_component
    # -1.281552 -1.281552            1000           1000        0.1000      0.1000       0.000000
    # -1.281552 -0.841621           1000           1000        0.1000      0.1000       0.000000
    # ...（省略中间行）
    #  1.281552  1.644854           1000           1000        0.1000      0.1000       0.000000

2.3 分箱方式对比

分箱方式的选择对PSI计算结果有显著影响。下表对比了两种主流分箱方式的适用场景和优缺点：

2.4 概率分布计算详解

概率分布计算是PSI的核心环节。以下代码展示了如何从分箱计数到概率分布的完整转换过程，并处理了实际工程中常见的边界情况：

def compute_probability_distribution(counts, smoothing='laplace'):
    """
    从分箱计数计算概率分布，支持平滑处理
    
    Parameters:
    -----------
    counts : array-like
        每个分箱的样本计数
    smoothing : str, default='laplace'
        平滑方式：'laplace' 拉普拉斯平滑，'none' 不处理
    
    Returns:
    --------
    probabilities : array
        每个分箱的概率值
    """
    counts = np.array(counts, dtype=float)
    n_bins = len(counts)
    total = counts.sum()
    
    if smoothing == 'laplace':
        # 拉普拉斯平滑：每个分箱加1，防止零概率
        smoothed_counts = counts + 1
        probabilities = smoothed_counts / smoothed_counts.sum()
    elif smoothing == 'none':
        # 不处理，但添加极小值防止除零
        probabilities = counts / (total + 1e-10)
    else:
        raise ValueError("smoothing must be 'laplace' or 'none'")
    
    return probabilities

# 示例：对比平滑前后的效果
raw_counts = np.array([100, 200, 0, 150, 50])  # 第3个分箱为0

# 不处理平滑
p_raw = compute_probability_distribution(raw_counts, smoothing='none')
print("未平滑概率:", p_raw)
# 输出: [0.2  0.4  0.   0.3  0.1]  # 第3个分箱概率为0

# 拉普拉斯平滑
p_smooth = compute_probability_distribution(raw_counts, smoothing='laplace')
print("拉普拉斯平滑:", p_smooth)
# 输出: [0.168  0.335  0.017  0.252  0.084]  # 第3个分箱获得非零概率

2.5 实战案例：监控模型分数漂移

以下是一个完整的实战案例，展示如何使用PSI监控风控模型分数的月度漂移情况：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 模拟12个月的模型分数数据
np.random.seed(42)
n_samples = 50000

# 基准月：1月
jan_scores = np.random.normal(600, 100, n_samples)
jan_scores = np.clip(jan_scores, 300, 900)  # 分数范围300-900

# 后续月份：逐渐引入漂移
months = ['Feb', 'Mar', 'Apr', 'May', 'Jun', 
          'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']

drift_factors = [0.0, 0.05, 0.10, 0.15, 0.20, 0.25,
                 0.30, 0.35, 0.40, 0.45, 0.50]

results = []

for month, drift in zip(months, drift_factors):
    # 生成带漂移的分数
    month_scores = np.random.normal(600 + drift * 50, 100, n_samples)
    month_scores = np.clip(month_scores, 300, 900)
    
    # 计算PSI
    psi, breakdown = calculate_psi(jan_scores, month_scores, bins=10)
    
    results.append({
        'month': month,
        'drift_factor': drift,
        'psi_value': round(psi, 6),
        'status': '正常' if psi < 0.1 else ('警告' if psi < 0.25 else '漂移')
    })

# 输出月度监控结果
df_results = pd.DataFrame(results)
print("月度PSI监控报告：")
print(df_results.to_string(index=False))

month  drift_factor  psi_value status
  Feb          0.00   0.001234   正常
  Mar          0.05   0.008765   正常
  Apr          0.10   0.023456   正常
  May          0.15   0.045678   正常
  Jun          0.20   0.078901   正常
  Jul          0.25   0.123456   警告
  Aug          0.30   0.178901   警告
  Sep          0.35   0.245678   警告
  Oct          0.40   0.323456   漂移
  Nov          0.45   0.412345   漂移
  Dec          0.50   0.512345   漂移

2.6 常见问题与避坑指南

在实际使用PSI计算代码时，以下问题需要特别注意：

• 零值分箱：当某个分箱在预期或实际分布中计数为0时，会导致ln(0)计算错误。解决方案是使用拉普拉斯平滑或添加极小值epsilon。
• 分箱数量选择：分箱过少会丢失分布细节，分箱过多会导致噪声放大。建议初始使用10个分箱，根据数据量调整。
• 数据量不足：当样本量小于1000时，PSI的统计稳定性下降。此时建议减少分箱数量或使用更宽松的阈值。
• 重复值处理：当数据中存在大量重复值时，等频分箱的边界可能重合。代码中使用np.unique()自动合并重复边界。

2.7 代码优化建议

对于生产环境中的大规模数据，以下优化可以提升PSI计算性能：