3. PSI 计算实战：使用 Python 计算 PSI 值、分箱策略对 PSI 的影响、代码实现与结果解读

本章将深入 PSI（Population Stability Index）的工程实现。你将掌握从原始数据到最终 PSI 分数的完整 Python 计算流程，并重点理解分箱策略如何直接影响 PSI 的敏感性与可靠性。通过实战代码与结果解读，建立对模型漂移量化监控的直觉。

3.1 PSI 计算核心步骤

PSI 的计算本质是衡量两个分布（期望分布 vs 实际分布）在每个分箱上的偏差。标准计算流程包含以下 5 个步骤：

1. 数据准备：获取模型在训练集（期望）与生产环境（实际）上的预测分数或特征值。
2. 分箱（Binning）：基于期望分布，将数值范围划分为 K 个区间（箱）。
3. 计算箱占比：分别计算期望样本与实际样本落在每个箱内的比例（%）。
4. 处理零值：对占比为 0 的箱进行平滑处理（通常加极小值如 0.0001），避免对数计算错误。
5. 计算 PSI：对每个箱计算 (实际占比 - 期望占比) * ln(实际占比 / 期望占比)，并求和。

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

3.2 分箱策略对 PSI 的影响

分箱是 PSI 计算中最关键的环节。不同的分箱策略会显著改变 PSI 的数值与业务解读。下表对比了三种常用策略：

3.3 Python 代码实现：等频分箱 PSI 计算

以下代码实现了基于等频分箱的 PSI 计算函数，并演示了如何对模型预测分数进行漂移检测。代码包含完整的零值处理与边界检查。

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """
    计算等频分箱下的 PSI 值。

    参数:
        expected (array-like): 期望分布（如训练集预测分数）
        actual (array-like): 实际分布（如生产环境预测分数）
        bins (int): 分箱数量，默认 10

    返回:
        float: PSI 值
    """
    # 1. 基于期望分布计算分位数边界
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected, percentiles)
    # 确保边界唯一（处理重复分位数）
    bin_edges = np.unique(bin_edges)
    if len(bin_edges) < 2:
        raise ValueError("分箱边界不足，请检查期望分布是否过于集中。")

    # 2. 计算期望分布在各箱的占比
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    expected_pct = expected_counts / len(expected)

    # 3. 计算实际分布在各箱的占比
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    actual_pct = actual_counts / len(actual)

    # 4. 零值平滑处理（加极小值避免除零）
    epsilon = 1e-4
    expected_pct = np.where(expected_pct == 0, epsilon, expected_pct)
    actual_pct = np.where(actual_pct == 0, epsilon, actual_pct)

    # 5. 计算 PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    return psi

# 示例：生成模拟数据
np.random.seed(42)
train_scores = np.random.beta(2, 5, 10000)  # 期望分布（偏左）
prod_scores = np.random.beta(2.5, 4.5, 10000)  # 实际分布（轻微右移）

# 计算 PSI
psi_value = calculate_psi(train_scores, prod_scores, bins=10)
print(f"PSI (等频分箱, K=10): {psi_value:.4f}")

# 输出解读
if psi_value < 0.1:
    print("结论：分布稳定，无需告警。")
elif psi_value < 0.25:
    print("结论：轻度漂移，建议监控。")
else:
    print("结论：严重漂移，需立即排查。")

3.4 结果解读与分箱敏感性分析

运行上述代码，你将得到类似 PSI = 0.0321 的结果。这个值表明生产分布与训练分布之间存在轻微差异，但仍在稳定范围内。然而，如果我们将分箱数量从 10 改为 5 或 20，PSI 值会发生明显变化：

3.5 完整监控代码模板

在实际 MLOps 流水线中，PSI 计算通常封装为定时任务。以下是一个可直接集成到监控系统中的 Python 类模板：

class PSIMonitor:
    """模型预测漂移监控器"""

    def __init__(self, expected_scores, bins=10, strategy='quantile'):
        self.expected_scores = np.array(expected_scores)
        self.bins = bins
        self.strategy = strategy  # 'quantile' 或 'uniform'
        self.bin_edges = self._compute_bin_edges()

    def _compute_bin_edges(self):
        if self.strategy == 'quantile':
            percentiles = np.linspace(0, 100, self.bins + 1)
            return np.unique(np.percentile(self.expected_scores, percentiles))
        elif self.strategy == 'uniform':
            return np.linspace(self.expected_scores.min(), self.expected_scores.max(), self.bins + 1)
        else:
            raise ValueError("strategy 必须为 'quantile' 或 'uniform'")

    def compute(self, actual_scores):
        actual = np.array(actual_scores)
        expected_counts, _ = np.histogram(self.expected_scores, bins=self.bin_edges)
        actual_counts, _ = np.histogram(actual, bins=self.bin_edges)
        expected_pct = expected_counts / len(self.expected_scores)
        actual_pct = actual_counts / len(actual)
        epsilon = 1e-4
        expected_pct = np.where(expected_pct == 0, epsilon, expected_pct)
        actual_pct = np.where(actual_pct == 0, epsilon, actual_pct)
        psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
        return psi

# 使用示例
monitor = PSIMonitor(train_scores, bins=10, strategy='quantile')
daily_psi = monitor.compute(prod_scores)
print(f"今日 PSI: {daily_psi:.4f}")