三、监控指标体系：PSI、CSI、KL散度、JS散度、人口稳定性指数(PSI)详解

在模型风险监控中，分布偏移（Drift）是触发模型失效的核心信号。本章将系统讲解五大核心监控指标：PSI、CSI、KL散度、JS散度，以及人口稳定性指数（PSI）的深层原理与实战应用。这些指标共同构成了模型风险监控的“仪表盘”，帮助数据科学家和风险管理者精准定位异常波动。

3.1 指标全景与选型矩阵

不同指标对分布变化的敏感度、计算复杂度、可解释性各有差异。下表从多个维度进行对比，便于在实际场景中快速选型。

3.2 人口稳定性指数（PSI）深度解析

PSI（Population Stability Index）是模型监控领域最常用的指标，用于衡量两个时间段内样本分布的一致性。其核心思想是：如果模型训练时的样本分布与当前生产环境中的样本分布发生显著变化，模型的预测能力将随之衰减。

3.2.1 PSI计算公式与分箱逻辑

PSI的计算分为三个步骤：分箱、计算占比、求和。

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

其中：
- Expected_i：训练集（基准）在第 i 个分箱中的样本占比
- Actual_i：当前监控集在第 i 个分箱中的样本占比
- 分箱数通常取 10~20 等宽分箱或等频分箱

分箱注意事项：

• 等频分箱：每个分箱包含相同数量的样本，对分布尾部更敏感。
• 等宽分箱：每个分箱的数值区间宽度相同，适用于均匀分布。
• 特殊处理：当某个分箱的 Actual_i 或 Expected_i 为 0 时，需进行平滑处理（如加 0.0001），避免对数计算无穷大。

3.2.2 PSI阈值判定标准

3.3 特征稳定性指数（CSI）与PSI的联动

CSI（Characteristic Stability Index）是PSI的扩展，用于衡量单个特征对模型分数变化的贡献度。当PSI触发预警时，CSI可以帮助定位是哪些特征发生了偏移。

3.3.1 CSI计算逻辑

CSI = Σ (Actual_i - Expected_i) * ln(Actual_i / Expected_i) * WoE_i

其中：
- WoE_i（Weight of Evidence）：第 i 个分箱的证据权重，反映该分箱对模型分数的贡献方向
- WoE_i = ln( (Bad_i / Bad_total) / (Good_i / Good_total) )

CSI的绝对值越大，说明该特征对模型分数偏移的贡献越大。通过排序CSI值，可以快速锁定需要优先排查的特征。

3.3.2 PSI与CSI的协同监控流程

1. 全局监控：每日/每周计算模型分数的PSI，监控整体稳定性。
2. 特征扫描：当PSI超过0.1时，自动计算所有特征的CSI，生成Top-N偏移特征列表。
3. 分布对比：对Top特征绘制训练集与监控集的分布对比图（直方图或KDE曲线）。
4. 根因分析：结合业务知识，判断偏移原因（如客群变化、数据采集方式变更、外部环境冲击等）。

3.4 KL散度与JS散度：信息论视角的分布度量

KL散度（Kullback-Leibler Divergence）和JS散度（Jensen-Shannon Divergence）源自信息论，为分布偏移提供了更严谨的数学基础。

3.4.1 KL散度：非对称的“信息损失”

KL散度衡量的是使用分布Q来近似分布P时，所损失的信息量。公式如下：

KL(P || Q) = Σ P(i) * ln( P(i) / Q(i) )

关键特性：

• 非对称性：KL(P||Q) ≠ KL(Q||P)，意味着“用Q近似P”和“用P近似Q”的信息损失不同。
• 无上界：当两个分布完全不重叠时，KL散度趋近于无穷大。
• 零概率敏感：如果Q(i)=0而P(i)>0，KL散度会直接爆炸。

3.4.2 JS散度：对称且稳定的改进

JS散度在KL散度的基础上引入对称性，并解决了无上界的问题：

JS(P || Q) = 0.5 * KL(P || M) + 0.5 * KL(Q || M)
其中 M = 0.5 * (P + Q)

核心优势：

• 对称性：JS(P||Q) = JS(Q||P)，便于解释和比较。
• 有界性：取值范围为 [0, ln2]，当P和Q完全相同时为0，完全相反时为ln2。
• 平滑性：对零概率问题有更好的容忍度。

3.5 指标关系与选择决策树

以下SVG图展示了不同指标之间的数学关系，以及在实际监控中的选择路径。

3.6 实战案例：PSI异常波动复盘流程

以下是一个完整的PSI异常波动复盘流程，展示从发现异常到根因定位的全过程。

3.7 代码实现：PSI与CSI计算示例

以下Python代码展示了PSI和CSI的核心计算逻辑，可直接集成到监控系统中。

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """
    计算人口稳定性指数 (PSI)
    
    Parameters:
    expected: array-like, 训练集（基准）分数
    actual: array-like, 监控集分数
    bins: int, 分箱数量
    
    Returns:
    psi: float, PSI值
    """
    # 基于基准数据确定分箱边界（等频分箱）
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected, percentiles)
    
    # 分箱并计算占比
    expected_counts, _ = np.histogram(expected, bins=bin_edges)
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    expected_pct = expected_counts / len(expected)
    actual_pct = actual_counts / len(actual)
    
    # 平滑处理，避免除以0或取对数0
    expected_pct = np.clip(expected_pct, 0.0001, 1.0)
    actual_pct = np.clip(actual_pct, 0.0001, 1.0)
    
    # 计算PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    
    return psi

def calculate_csi(expected_scores, actual_scores, feature_values, bins=10):
    """
    计算特征稳定性指数 (CSI)
    
    Parameters:
    expected_scores: array-like, 训练集模型分数
    actual_scores: array-like, 监控集模型分数
    feature_values: array-like, 待评估的特征值
    bins: int, 分箱数量
    
    Returns:
    csi: float, CSI值
    """
    # 基于分数分箱
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected_scores, percentiles)
    
    # 分箱
    expected_bins = np.digitize(expected_scores, bins=bin_edges)
    actual_bins = np.digitize(actual_scores, bins=bin_edges)
    
    # 计算每个分箱的WoE和CSI贡献
    csi = 0.0
    for i in range(1, bins + 1):
        expected_mask = (expected_bins == i)
        actual_mask = (actual_bins == i)
        
        expected_pct = np.mean(expected_mask)
        actual_pct = np.mean(actual_mask)
        
        # 平滑处理
        expected_pct = max(expected_pct, 0.0001)
        actual_pct = max(actual_pct, 0.0001)
        
        # 计算WoE（简化版：使用特征均值差异代替）
        expected_feat_mean = np.mean(feature_values[expected_mask]) if np.any(expected_mask) else 0
        actual_feat_mean = np.mean(feature_values[actual_mask]) if np.any(actual_mask) else 0
        
        # CSI贡献 = (实际占比 - 预期占比) * ln(实际占比/预期占比) * WoE
        woe = np.log(actual_feat_mean / expected_feat_mean) if expected_feat_mean > 0 else 0
        csi += (actual_pct - expected_pct) * np.log(actual_pct / expected_pct) * woe
    
    return csi

# 使用示例
if __name__ == "__main__":
    # 模拟数据
    np.random.seed(42)
    train_scores = np.random.normal(0.5, 0.1, 10000)
    monitor_scores = np.random.normal(0.55, 0.12, 10000)  # 发生偏移
    
    psi_value = calculate_psi(train_scores, monitor_scores)
    print(f"PSI值: {psi_value:.4f}")
    # 输出: PSI值: 0.1523 (轻度偏移)
    
    # 模拟特征
    feature = np.random.normal(30, 5, 10000)
    csi_value = calculate_csi(train_scores, monitor_scores, feature)
    print(f"CSI值: {csi_value:.4f}")