2. 特征漂移基础：什么是特征漂移（数据漂移）、漂移的类型

在贷后预警体系中，模型上线后并非一成不变。随着宏观经济、客群行为、业务策略的变化，模型输入数据的统计分布或输入与目标之间的关系会发生改变，这种现象统称为特征漂移（Feature Drift）或数据漂移（Data Drift）。若不及时发现并处理漂移，模型预测能力将快速衰减，导致预警失效。

2.1 什么是特征漂移（数据漂移）

特征漂移是指模型在生产环境中接收到的实时数据分布，与模型训练时所使用的历史数据分布之间出现显著差异。这种差异会破坏模型“训练分布≈实时分布”的基本假设，使模型输出偏离预期。

漂移的量化通常通过统计检验（如 KS 检验、PSI 群体稳定性指标）或距离度量（如 Wasserstein 距离）来评估。下表列出了常用检测指标及其适用场景：

2.2 漂移的三种类型

根据漂移发生的对象不同，学术界和工业界将其划分为三类：协变量漂移、先验概率漂移和概念漂移。理解这三者的区别，是制定告警阈值和应对策略的前提。

2.2.1 协变量漂移（Covariate Shift）

协变量漂移指模型输入特征 X 的边际分布 P(X) 发生变化，但条件分布 P(Y|X) 保持不变。即“客群结构变了，但相同特征下的违约规律没变”。

• 典型场景： 疫情期间，申请客群的年龄、收入分布发生偏移，但相同收入水平下的违约率关系不变。
• 检测方法： 对每个特征单独计算 PSI 或 KS 统计量。
• 应对策略： 重采样、特征标准化、模型重训练。

2.2.2 先验概率漂移（Prior Probability Shift）

先验概率漂移指目标变量 Y 的边际分布 P(Y) 发生变化，但条件分布 P(X|Y) 保持不变。即“整体违约率变了，但各类别内部的特征分布没变”。

• 典型场景： 经济下行期，整体逾期率从 2% 上升至 8%，但好客户和坏客户的特征模式未变。
• 检测方法： 监控模型输出的平均预测概率或坏样本占比。
• 应对策略： 调整决策阈值、重新校准模型概率。

2.2.3 概念漂移（Concept Drift）

概念漂移指输入 X 与目标 Y 之间的映射关系 P(Y|X) 发生变化。这是最危险的一类漂移，意味着“过去的规律已经失效”。

• 典型场景： 监管政策变化后，过去“高负债=高风险”的规律变为“高负债但稳定现金流=低风险”。
• 检测方法： 监控模型残差、滚动窗口 AUC 衰减、PSI 与模型性能联合分析。
• 应对策略： 必须重训练或重建模型，仅靠阈值调整无效。

2.3 三类漂移的对比总结

下表从多个维度对三种漂移进行对比，帮助在实际监控中快速定位问题类型：

2.4 漂移检测的实践流程

在贷后预警系统中，建议按以下步骤建立漂移检测流水线：

1. 特征级监控： 对每个入模特征，按日/周计算 PSI，标记超过阈值的特征。
2. 模型级监控： 监控模型输出的平均预测概率、分数分布、AUC 等指标。
3. 联合诊断： 当特征漂移与性能下降同时发生时，进一步分析是否属于概念漂移。
4. 告警分级： 根据漂移类型和严重程度，触发不同级别的告警通知。

2.5 漂移检测的代码示例

以下 Python 代码演示如何使用 PSI 检测单个特征的漂移：

import numpy as np
import pandas as pd

def calculate_psi(expected, actual, bins=10):
    """
    计算群体稳定性指标 (PSI)
    :param expected: 训练集特征值 (array-like)
    :param actual: 实时特征值 (array-like)
    :param bins: 分箱数
    :return: PSI 值
    """
    # 基于训练集确定分箱边界
    percentiles = np.linspace(0, 100, bins + 1)
    bin_edges = np.percentile(expected, percentiles)
    
    # 对两个分布进行分箱
    expected_binned = np.histogram(expected, bins=bin_edges)[0] + 1e-10
    actual_binned = np.histogram(actual, bins=bin_edges)[0] + 1e-10
    
    # 计算占比
    expected_pct = expected_binned / expected_binned.sum()
    actual_pct = actual_binned / actual_binned.sum()
    
    # 计算 PSI
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    return psi

# 示例：检测特征 'income' 的漂移
train_income = np.random.normal(50000, 15000, 10000)
real_time_income = np.random.normal(55000, 18000, 1000)  # 分布偏移

psi_value = calculate_psi(train_income, real_time_income)
print(f"特征 income 的 PSI 值为: {psi_value:.4f}")

if psi_value > 0.25:
    print("告警：严重漂移，需立即排查")
elif psi_value > 0.1:
    print("提示：轻度漂移，建议关注")
else:
    print("正常：无显著漂移")