3. PSI计算步骤详解：分箱策略、分布占比、对数变换与求和

3.5 Python代码实现

以下代码实现了完整的PSI计算函数，包含空箱平滑处理：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10, method='quantile', epsilon=1e-6):
    """
    计算群体稳定性指标（PSI）
    
    Parameters:
    -----------
    expected : array-like
        基准样本分数
    actual : array-like
        当前样本分数
    bins : int
        分箱数量（等频分箱时有效）
    method : str
        分箱方法：'quantile'（等频）, 'equal_width'（等距）
    epsilon : float
        平滑因子，防止除以0或ln(0)
    
    Returns:
    --------
    psi : float
        PSI值
    bin_df : DataFrame
        每箱的详细计算过程
    """
    # 1. 确定分箱边界（基于基准样本）
    if method == 'quantile':
        # 等频分箱：使用基准样本的分位数
        percentiles = np.linspace(0, 100, bins + 1)
        # 处理极端值：使用1%和99%分位数作为边界
        low = np.percentile(expected, 1)
        high = np.percentile(expected, 99)
        # 在[low, high]范围内计算分位数
        edges = np.percentile(
            np.clip(expected, low, high), 
            percentiles[1:-1]
        )
        edges = np.concatenate([[low], edges, [high + 1]])
    elif method == 'equal_width':
        # 等距分箱
        low = np.min(expected)
        high = np.max(expected)
        edges = np.linspace(low, high + 1, bins + 1)
    else:
        raise ValueError("method must be 'quantile' or 'equal_width'")
    
    # 2. 计算每箱样本数
    expected_counts, _ = np.histogram(expected, bins=edges)
    actual_counts, _ = np.histogram(actual, bins=edges)
    
    # 3. 计算占比（加平滑）
    expected_pct = (expected_counts + epsilon) / (np.sum(expected_counts) + epsilon * bins)
    actual_pct = (actual_counts + epsilon) / (np.sum(actual_counts) + epsilon * bins)
    
    # 4. 计算单箱贡献
    ratio = actual_pct / expected_pct
    contribution = (actual_pct - expected_pct) * np.log(ratio)
    
    # 5. 求和得到PSI
    psi = np.sum(contribution)
    
    # 6. 构建结果DataFrame
    bin_df = pd.DataFrame({
        'bin_lower': edges[:-1],
        'bin_upper': edges[1:],
        'expected_count': expected_counts,
        'actual_count': actual_counts,
        'expected_pct': expected_pct,
        'actual_pct': actual_pct,
        'ratio': ratio,
        'log_ratio': np.log(ratio),
        'contribution': contribution
    })
    
    return psi, bin_df

# 使用示例
np.random.seed(42)
expected_scores = np.random.normal(600, 100, 10000)
actual_scores = np.random.normal(580, 120, 10000)  # 模拟漂移

psi_value, detail_df = calculate_psi(expected_scores, actual_scores, bins=10)
print(f"PSI = {psi_value:.4f}")
print(detail_df.head())

3.6 常见计算陷阱与规避

掌握以上四步计算流程和工程实现细节，即可在生产环境中稳定、准确地计算PSI，为后续的漂移告警阈值设定和自动化监控奠定基础。