📚 课程目录第3章 PSI计算实战（一）

3. PSI计算实战（一）：使用Python计算单个特征的PSI值，数据准备与分箱逻辑详解

在构建批次偏移监控系统时，PSI（Population Stability Index，群体稳定性指标）是最常用的量化指标之一。本章将聚焦于单个特征的PSI计算实战，从数据准备到分箱逻辑，逐步拆解计算过程。理解这些基础操作，是后续实现自动化监控与阈值设置的前提。

3.1 PSI计算的核心流程

PSI的计算本质上是比较两个分布（基准分布 vs. 当前分布）在每个分箱中的占比差异。下图展示了从原始数据到最终PSI值的完整计算链路：

3.2 数据准备：基准集与当前集

PSI计算需要两个数据集：

基准集（Expected）：通常选取模型训练集、验证集或上线初期的稳定批次，作为分布参考标准。
当前集（Actual）：待监控的新批次数据，用于与基准集比较偏移程度。

在Python中，我们通常将数据组织为两个Series或DataFrame列。以下是一个典型的数据准备示例：

import pandas as pd
import numpy as np

# 模拟基准集（训练集）特征
np.random.seed(42)
baseline = pd.Series(np.random.normal(loc=50, scale=10, size=10000), name='feature_a')

# 模拟当前集（新批次）特征，均值偏移 +5
current = pd.Series(np.random.normal(loc=55, scale=10, size=10000), name='feature_a')

print(f"基准集样本数: {len(baseline)}")
print(f"当前集样本数: {len(current)}")
print(f"基准集均值: {baseline.mean():.2f}, 当前集均值: {current.mean():.2f}")

实践建议：基准集与当前集的样本量应尽量接近，避免因样本量差异过大导致占比计算失真。若当前集样本量远小于基准集，建议使用分层抽样或调整分箱数量。

3.3 分箱逻辑详解

分箱是PSI计算中最关键的步骤。分箱方式直接影响PSI值的敏感性与稳定性。下表对比了三种主流分箱策略：

分箱策略	原理	适用场景	优点	缺点
等距分箱	将特征值范围均匀划分为K个区间	特征分布均匀，无极端异常值	实现简单，边界固定	对稀疏区间敏感，可能导致空箱
等频分箱	使每个箱包含大致相同的样本数	特征分布偏态或含异常值	避免空箱，对分布鲁棒	边界随数据变化，跨批次可比性差
自定义分箱	基于业务知识或模型分箱点	评分卡、信用评分等业务场景	业务可解释性强	需要领域知识，维护成本高

3.4 等距分箱实战代码

等距分箱是最常用的方法。以下代码演示如何基于基准集确定分箱边界，并应用到当前集：

def psi_equidistant(baseline, current, bins=10):
    """
    等距分箱计算PSI
    """
    # 1. 基于基准集确定分箱边界
    min_val = baseline.min()
    max_val = baseline.max()
    bin_edges = np.linspace(min_val, max_val, bins + 1)
    
    # 2. 对基准集和当前集分别分箱
    baseline_bins = np.clip(np.digitize(baseline, bin_edges), 1, bins)
    current_bins = np.clip(np.digitize(current, bin_edges), 1, bins)
    
    # 3. 计算每个箱的占比
    baseline_counts = np.bincount(baseline_bins, minlength=bins+1)[1:]
    current_counts = np.bincount(current_bins, minlength=bins+1)[1:]
    
    baseline_pct = baseline_counts / len(baseline)
    current_pct = current_counts / len(current)
    
    # 4. 处理零占比（加极小值避免除零）
    baseline_pct = np.where(baseline_pct == 0, 1e-6, baseline_pct)
    current_pct = np.where(current_pct == 0, 1e-6, current_pct)
    
    # 5. 计算PSI
    psi = np.sum((baseline_pct - current_pct) * np.log(baseline_pct / current_pct))
    
    return psi, bin_edges

# 计算PSI
psi_value, edges = psi_equidistant(baseline, current, bins=10)
print(f"等距分箱 PSI = {psi_value:.6f}")
print(f"分箱边界: {np.round(edges, 2)}")

3.5 等频分箱实战代码

等频分箱能更好地处理偏态分布。注意：分箱边界必须由基准集确定，并固定用于当前集：

def psi_equifrequency(baseline, current, bins=10):
    """
    等频分箱计算PSI
    """
    # 1. 基于基准集计算分位数边界
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(baseline, percentiles)
    # 确保边界唯一
    bin_edges = np.unique(bin_edges)
    
    # 2. 分箱
    baseline_bins = np.clip(np.digitize(baseline, bin_edges), 1, len(bin_edges)-1)
    current_bins = np.clip(np.digitize(current, bin_edges), 1, len(bin_edges)-1)
    
    # 3. 计算占比
    baseline_counts = np.bincount(baseline_bins, minlength=len(bin_edges))[1:]
    current_counts = np.bincount(current_bins, minlength=len(bin_edges))[1:]
    
    baseline_pct = baseline_counts / len(baseline)
    current_pct = current_counts / len(current)
    
    # 4. 平滑处理
    baseline_pct = np.where(baseline_pct == 0, 1e-6, baseline_pct)
    current_pct = np.where(current_pct == 0, 1e-6, current_pct)
    
    # 5. 计算PSI
    psi = np.sum((baseline_pct - current_pct) * np.log(baseline_pct / current_pct))
    
    return psi, bin_edges

psi_value_eqfreq, edges_eqfreq = psi_equifrequency(baseline, current, bins=10)
print(f"等频分箱 PSI = {psi_value_eqfreq:.6f}")

3.6 分箱策略对比与选择

不同分箱策略对同一组数据可能产生不同的PSI值。以下对比图展示了两种策略在相同数据上的差异：

3.7 分箱中的常见陷阱与处理

⚠ 空箱问题：当当前集某个箱内样本数为0时，占比Q_i=0，导致ln(P_i/0)无穷大。解决方案：对所有占比加极小值（如1e-6）进行平滑处理，或使用等频分箱避免空箱。

  🔑 边界一致性：分箱边界必须由基准集确定，并作为固定参数应用到所有后续批次。如果每次重新计算边界，PSI将失去跨批次可比性，无法反映真实的分布偏移。

3.8 完整PSI计算函数封装

将上述逻辑封装为通用函数，便于后续批量特征监控：

def calculate_psi(baseline, current, bins=10, bin_type='equidistant'):
    """
    通用PSI计算函数
    
    参数:
        baseline: 基准集 Series
        current: 当前集 Series
        bins: 分箱数量
        bin_type: 分箱类型，'equidistant' 或 'equifrequency'
    
    返回:
        psi: PSI值
        bin_edges: 分箱边界
    """
    if bin_type == 'equidistant':
        bin_edges = np.linspace(baseline.min(), baseline.max(), bins + 1)
    elif bin_type == 'equifrequency':
        percentiles = np.linspace(0, 100, bins + 1)
        bin_edges = np.percentile(baseline, percentiles)
        bin_edges = np.unique(bin_edges)
    else:
        raise ValueError("bin_type 必须为 'equidistant' 或 'equifrequency'")
    
    # 分箱
    baseline_bins = np.clip(np.digitize(baseline, bin_edges), 1, len(bin_edges)-1)
    current_bins = np.clip(np.digitize(current, bin_edges), 1, len(bin_edges)-1)
    
    # 占比计算
    baseline_counts = np.bincount(baseline_bins, minlength=len(bin_edges))[1:]
    current_counts = np.bincount(current_bins, minlength=len(bin_edges))[1:]
    
    baseline_pct = baseline_counts / len(baseline)
    current_pct = current_counts / len(current)
    
    # 平滑
    baseline_pct = np.where(baseline_pct == 0, 1e-6, baseline_pct)
    current_pct = np.where(current_pct == 0, 1e-6, current_pct)
    
    # PSI
    psi = np.sum((baseline_pct - current_pct) * np.log(baseline_pct / current_pct))
    
    return psi, bin_edges

# 测试
psi_eq, _ = calculate_psi(baseline, current, bins=10, bin_type='equidistant')
psi_ef, _ = calculate_psi(baseline, current, bins=10, bin_type='equifrequency')
print(f"等距PSI: {psi_eq:.6f} | 等频PSI: {psi_ef:.6f}")