风控模型稳定性监控：PSI 与 KS 指标实战

模型上线不是终点，监控才是风控模型真正的开始。

一、为什么需要模型稳定性监控

信贷风控模型上线后，最怕的不是模型不准，而是模型悄悄变差你还不知道。

一个典型的翻车场景：评分卡上线时 KS=0.35，半年后坏账率突然飙升。排查发现，经济下行导致客群分布发生漂移——原来评分 600 分以上算好客户，现在 650 分以下的都在逾期。模型没变，但世界变了。

这就是模型稳定性监控要解决的问题。核心回答两个问题：

1. 客群漂移了吗？（PSI — Population Stability Index）
2. 模型区分能力还在吗？（KS — Kolmogorov-Smirnov）

二、PSI：人群稳定性指标

2.1 原理

PSI 衡量的是建模样本和线上样本的分布差异。说白了就是：现在进件的这批人，跟当初建模那批人长得还一样吗？

计算公式：

$$ PSI = \sum_{i=1}^{n} (A_i - E_i) \times \ln\frac{A_i}{E_i} $$

• $A_i$（Actual）：线上样本在第 i 个分箱中的占比
• $E_i$（Expected）：建模样本在第 i 个分箱中的占比

判稳标准：

PSI 值	结论
< 0.1	分布稳定，无需关注
0.1 ~ 0.25	轻微漂移，需观察
> 0.25	显著漂移，必须排查

2.2 代码实现

import numpy as np
import pandas as pd

def calc_psi(expected, actual, bins=10):
    """
    计算 PSI
    expected: 建模样本的评分分布 (pd.Series)
    actual: 线上样本的评分分布 (pd.Series)
    bins: 分箱数
    """
    # 基于 expected 的分布确定分箱边界
    breaks = np.percentile(expected, np.linspace(0, 100, bins + 1))
    breaks = np.unique(breaks)  # 去重，防止重复边界

    # 统计各箱占比
    expected_cnt = pd.cut(expected, breaks, include_lowest=True).value_counts()
    actual_cnt = pd.cut(actual, breaks, include_lowest=True).value_counts()

    expected_pct = expected_cnt / expected_cnt.sum()
    actual_pct = actual_cnt / actual_cnt.sum()

    # 防止除零：将 0 替换为极小值
    expected_pct = expected_pct.replace(0, 1e-10)
    actual_pct = actual_pct.replace(0, 1e-10)

    psi = ((actual_pct - expected_pct) * np.log(actual_pct / expected_pct)).sum()
    return round(psi, 4)


# ===== 使用示例 =====
# 模拟数据：建模样本 vs 线上样本的评分
np.random.seed(42)
expected_score = np.random.normal(600, 50, 10000)    # 建模时：均值 600
actual_score_ok = np.random.normal(605, 55, 10000)    # 正常：略偏但稳定
actual_score_bad = np.random.normal(560, 70, 10000)   # 异常：均值漂移 40 分

print(f"正常线上 PSI: {calc_psi(expected_score, actual_score_ok)}")   # ~0.05
print(f"异常线上 PSI: {calc_psi(expected_score, actual_score_bad)}")  # ~0.35

2.3 多维度 PSI

实际工作中，不仅要看总分的 PSI，还要拆到特征级 PSI。某个特征漂移可能是总分漂移的根因。

def calc_feature_psi(df_train, df_online, feature_cols, bins=10):
    """逐特征计算 PSI"""
    psi_result = {}
    for col in feature_cols:
        psi_result[col] = calc_psi(
            df_train[col].dropna(),
            df_online[col].dropna(),
            bins=bins
        )
    return pd.Series(psi_result).sort_values(ascending=False)


# 示例
features = ['age', 'income', 'credit_history_length', 'loan_amount']
feature_psi = calc_feature_psi(df_train, df_online, features)
print(feature_psi)
# 输出每个特征的 PSI，最高的那个优先排查

三、KS：模型区分能力指标

3.1 原理

KS 衡量模型区分好坏客户的能力。它的计算过程是：

1. 将样本按模型分从低到高排序
2. 在每个分界点，计算累计好客户占比和累计坏客户占比
3. KS = max(|累计好客户占比 - 累计坏客户占比|)

KS 值越大，模型区分力越强。风控模型常用判标：

KS 值	结论
< 0.2	区分力差，不建议上线
0.2 ~ 0.3	可用，有区分力
0.3 ~ 0.5	良好，区分力较强
> 0.5	优秀，但需检查过拟合
> 0.75	几乎肯定数据泄漏

3.2 代码实现

import matplotlib.pyplot as plt
from sklearn.metrics import roc_curve

def calc_ks(y_true, y_pred):
    """
    计算 KS 值
    y_true: 真实标签 (0=好, 1=坏)
    y_pred: 模型预测概率
    """
    fpr, tpr, thresholds = roc_curve(y_true, y_pred)
    ks = max(tpr - fpr)
    return round(ks, 4)


def plot_ks_curve(y_true, y_pred):
    """绘制 KS 曲线"""
    fpr, tpr, thresholds = roc_curve(y_true, y_pred)
    ks = max(tpr - fpr)
    ks_idx = np.argmax(tpr - fpr)

    plt.figure(figsize=(10, 6))
    plt.plot(thresholds, tpr, label='累计坏客户占比 (TPR)', linewidth=2)
    plt.plot(thresholds, fpr, label='累计好客户占比 (FPR)', linewidth=2)
    plt.plot(thresholds, tpr - fpr, label='KS 曲线 (TPR-FPR)', linewidth=2, linestyle='--')
    plt.axvline(thresholds[ks_idx], color='red', linestyle=':',
                label=f'最大 KS = {ks:.4f} (阈值={thresholds[ks_idx]:.4f})')
    plt.xlabel('阈值')
    plt.ylabel('累计占比')
    plt.title('KS 曲线')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.show()

    return ks


# ===== 使用示例 =====
y_true = np.random.binomial(1, 0.1, 10000)
y_pred_good = np.random.beta(2, 8, 10000)       # 区分力好
y_pred_bad = np.random.beta(4, 6, 10000)         # 区分力差

print(f"好模型 KS: {calc_ks(y_true, y_pred_good)}")  # ~0.35
print(f"差模型 KS: {calc_ks(y_true, y_pred_bad)}")   # ~0.10

3.3 KS 下降排查思路

当监控发现 KS 下降，排查路径如下：

1. 特征缺失/空值率飙升 → 上游数据源异常
2. 某个关键变量 PSI 飙升 → 客群结构变化
3. 标签回传延迟 → 坏样本还没暴露，分母虚高
4. 策略/规则调整 → 高分段客户被新的准入规则拒绝，评分分布被截断
5. 数据泄漏修复 → 之前 KS 虚高是数据泄漏，现在回归真实水平

四、PSI + KS 联合监控：生产级方案

单独的 PSI 或 KS 都不够。真正靠谱的做法是双指标联动 + 自动化监控。

4.1 监控 SQL（每日自动跑）

假设评分结果落库在 model_score_daily 表：

-- ===== PSI 监控：按天对比基准分布 =====
-- 基准：建模样本的评分分箱占比（存为常量或参数表）
WITH base AS (
    SELECT 'low'    AS bucket, 0.15 AS base_pct UNION ALL   -- 0-550 分
    SELECT 'mid_l'  AS bucket, 0.25 AS base_pct UNION ALL   -- 550-600
    SELECT 'mid'    AS bucket, 0.30 AS base_pct UNION ALL   -- 600-650
    SELECT 'mid_h'  AS bucket, 0.20 AS base_pct UNION ALL   -- 650-700
    SELECT 'high'   AS bucket, 0.10 AS base_pct             -- 700+
),
today AS (
    SELECT
        CASE
            WHEN score < 550 THEN 'low'
            WHEN score < 600 THEN 'mid_l'
            WHEN score < 650 THEN 'mid'
            WHEN score < 700 THEN 'mid_h'
            ELSE 'high'
        END AS bucket,
        COUNT(*) * 1.0 / SUM(COUNT(*)) OVER() AS actual_pct
    FROM model_score_daily
    WHERE dt = '${yesterday}'
    GROUP BY 1
)
SELECT
    '${yesterday}' AS monitor_date,
    ROUND(SUM((t.actual_pct - b.base_pct) * LN(t.actual_pct / b.base_pct)), 4) AS psi,
    CASE
        WHEN SUM((t.actual_pct - b.base_pct) * LN(t.actual_pct / b.base_pct)) < 0.1 THEN '正常'
        WHEN SUM((t.actual_pct - b.base_pct) * LN(t.actual_pct / b.base_pct)) < 0.25 THEN '预警'
        ELSE '告警'
    END AS psi_status
FROM today t
JOIN base b ON t.bucket = b.bucket;


-- ===== KS 监控：按天计算 KS =====
SELECT
    dt,
    MAX(CASE WHEN label = 1 THEN cum_bad_pct ELSE cum_good_pct END
        - CASE WHEN label = 0 THEN cum_good_pct ELSE cum_bad_pct END) AS ks,
    CASE
        WHEN MAX(...) >= 0.30 THEN '正常'
        WHEN MAX(...) >= 0.20 THEN '预警'
        ELSE '告警'
    END AS ks_status
FROM (
    SELECT
        dt, score, label,
        SUM(CASE WHEN label = 1 THEN 1 ELSE 0 END)
            OVER (ORDER BY score) * 1.0
            / SUM(CASE WHEN label = 1 THEN 1 ELSE 0 END) OVER () AS cum_bad_pct,
        SUM(CASE WHEN label = 0 THEN 1 ELSE 0 END)
            OVER (ORDER BY score) * 1.0
            / SUM(CASE WHEN label = 0 THEN 1 ELSE 0 END) OVER () AS cum_good_pct
    FROM model_score_daily
    WHERE dt = '${yesterday}'
) t
GROUP BY dt;

4.2 自动化告警脚本

import smtplib
from email.mime.text import MIMEText
from datetime import datetime, timedelta


class ModelMonitor:
    """风控模型自动化监控器"""

    def __init__(self, psi_threshold=0.25, ks_threshold=0.20):
        self.psi_threshold = psi_threshold
        self.ks_threshold = ks_threshold

    def monitor(self, psi, ks, feature_psi=None):
        """综合判断模型健康状态"""
        alerts = []
        level = '正常'

        # 1. 总分 PSI 检查
        if psi > 0.25:
            alerts.append(f'[严重] 总分 PSI={psi:.4f}，客群分布显著漂移')
            level = '严重'
        elif psi > 0.10:
            alerts.append(f'[预警] 总分 PSI={psi:.4f}，客群轻微漂移')
            level = max(level, '预警')

        # 2. KS 检查
        if ks < 0.15:
            alerts.append(f'[严重] KS={ks:.4f}，模型区分力丧失')
            level = '严重'
        elif ks < 0.20:
            alerts.append(f'[预警] KS={ks:.4f}，区分力下降')
            level = max(level, '预警')

        # 3. 特征级 PSI 检查
        if feature_psi is not None:
            high_psi_features = feature_psi[feature_psi > 0.15]
            for feat, val in high_psi_features.items():
                alerts.append(f'[预警] 特征 [{feat}] PSI={val:.4f}，建议排查')

        return {
            'date': datetime.now().strftime('%Y-%m-%d'),
            'psi': psi,
            'ks': ks,
            'level': level,
            'alerts': alerts,
            'action': self._suggest_action(level, alerts)
        }

    def _suggest_action(self, level, alerts):
        """根据告警等级给出建议动作"""
        actions = {
            '正常': '无需操作，继续日常监控',
            '预警': '建议：(1) 检查上游数据源 (2) 分析高 PSI 特征的业务含义 (3) 观察 3-5 个工作日是否持续恶化',
            '严重': '建议：(1) 立即排查数据源异常 (2) 评估是否需紧急回滚模型 (3) 通知业务方暂停自动审批'
        }
        return actions.get(level, '')


# ===== 每日监控任务 =====
monitor = ModelMonitor()

# 从数据库或日志拉取昨日指标
report = monitor.monitor(
    psi=0.12,
    ks=0.28,
    feature_psi=pd.Series({'age': 0.18, 'income': 0.05, 'loan_amount': 0.03})
)

print(f"日期: {report['date']}")
print(f"PSI: {report['psi']}, KS: {report['ks']}")
print(f"等级: {report['level']}")
for alert in report['alerts']:
    print(f"  {alert}")
print(f"建议: {report['action']}")

4.3 PSI + KS 四象限决策矩阵

两个指标交叉，形成四种典型场景：

场景	PSI	KS	含义	动作
🟢 一切正常	低	高	客群稳定，模型有效	继续监控
🟡 客群漂移	高	高	人变了但模型还能分	观察，必要时重建模型
🟡 模型衰退	低	低	人没变但区分力下降	排查特征/标签/过拟合
🔴 双重恶化	高	低	人变了+分不出了	立即重建模型

五、常见坑与最佳实践

坑 1：只看总分不看特征

总分 PSI 正常不代表特征级没问题。多个特征相互抵消可能导致总分 PSI 正常，但每个特征都在悄悄漂移。

做法：总分 PSI + 特征级 PSI 都要看。

坑 2：KS 高就万事大吉

KS 突然飙升到 0.6 以上，别高兴太早——大概率是数据泄漏（特征里混入了未来信息或标签信息）。

做法：KS > 0.5 自动触发特征审查，逐个检查是否有时间穿越。

坑 3：建模样本选错了

PSI 的 "Expected" 用什么样本？应该用建模时通过的样本，不是全量样本。如果建模时做了拒绝推断（Reject Inference），记得把拒绝推断的样本也纳入 Expected。

坑 4：只看日级不看趋势

某一天 PSI 飙到 0.3，可能是偶然波动（促销活动、节假日）。不要一看到告警就拉警报。

做法：用 7 日滑动平均 + 趋势线，连续 3 天超标才触发严重告警。

六、总结

风控模型监控的核心要点：

1. PSI 看分布：客群是不是变了？总分 + 特征两级都要看
2. KS 看能力：模型还能不能区分好坏？下降要追溯根因
3. 双指标联动：PSI 高 + KS 高 = 人变了但还能分；PSI 高 + KS 低 = 紧急重建
4. 自动化优先：SQL 每日跑指标 → Python 脚本判断告警 → 钉钉/企微/邮件推送
5. 趋势比单点重要：别被一天的波动吓到，看连续趋势

模型上线只是第一步，持续监控才是风控工程师真正的护城河。

---

下一篇预告：信贷风控中的样本不均衡处理 — SMOTE、欠采样与代价敏感学习