4. 数值特征漂移检测（KS检验）

在贷后预警场景中，数值型特征（如收入、负债率、逾期天数）的分布变化是模型失效的核心诱因。KS检验（Kolmogorov-Smirnov Test）是检测数值特征分布漂移的经典非参数方法，无需假设数据服从特定分布，直接比较两个样本的累积分布函数（CDF）差异。

4.1 KS检验原理

KS检验的核心思想：计算训练集（参考集）与生产集（当前集）的累积分布函数之间的最大垂直距离，该距离即为KS统计量。若KS统计量超过临界值，则认为两个样本来自不同分布，即发生特征漂移。

KS检验的假设体系：

• 原假设 H₀：两个样本来自同一连续分布（无漂移）
• 备择假设 H₁：两个样本来自不同连续分布（存在漂移）

KS统计量 D 的取值范围为 [0, 1]，值越大表示分布差异越显著。当 D 接近 0 时，两个分布几乎重合；当 D 接近 1 时，分布完全分离。

4.2 scipy.stats.ks_2samp 实现

Python 中通过 scipy.stats.ks_2samp 函数直接完成双样本 KS 检验。该函数返回 KS 统计量 D 和 p 值，无需手动计算 CDF。

函数签名：

scipy.stats.ks_2samp(data1, data2, alternative='two-sided', mode='auto')

参数说明：

贷后预警实战代码：

import pandas as pd
import numpy as np
from scipy.stats import ks_2samp

# 模拟数据：训练集 vs 生产集
np.random.seed(42)
train_income = np.random.lognormal(mean=10.5, sigma=0.6, size=5000)  # 训练集收入
prod_income = np.random.lognormal(mean=10.8, sigma=0.7, size=5000)  # 生产集收入（均值漂移）

# 执行KS检验
stat, p_value = ks_2samp(train_income, prod_income, alternative='two-sided')

print(f"KS统计量 D = {stat:.4f}")
print(f"p值 = {p_value:.6f}")

# 漂移判定
alpha = 0.05
if p_value < alpha:
    print("结论：拒绝H₀ → 特征分布发生显著漂移")
else:
    print("结论：无法拒绝H₀ → 特征分布稳定")

4.3 p值解读与阈值设定

p值表示在原假设为真（无漂移）的条件下，观察到当前KS统计量或更极端值的概率。p值越小，拒绝原假设的证据越强。

p值解读规则：

阈值设定建议：

• 严格阈值（α=0.01）：适用于高风险特征（如逾期天数、负债率），减少误报，避免频繁重训
• 标准阈值（α=0.05）：适用于大多数数值特征，平衡检测灵敏度与误报率
• 宽松阈值（α=0.10）：适用于探索性分析或样本量较小时，提高漂移检出率

4.4 实战：贷后特征漂移监控表

将KS检验集成到贷后预警流水线中，构建特征漂移监控表，实现自动化检测。

import pandas as pd
from scipy.stats import ks_2samp

def ks_drift_detection(ref_df, prod_df, features, alpha=0.05):
    """
    批量检测数值特征漂移
    ref_df: 训练集DataFrame
    prod_df: 生产集DataFrame
    features: 待检测特征列表
    alpha: 显著性水平
    """
    results = []
    for feat in features:
        stat, p_value = ks_2samp(ref_df[feat].dropna(),
                                 prod_df[feat].dropna())
        drift_flag = "漂移" if p_value < alpha else "稳定"
        results.append({
            "特征": feat,
            "KS统计量": round(stat, 4),
            "p值": round(p_value, 6),
            "判定": drift_flag
        })
    return pd.DataFrame(results)

# 示例调用
ref_data = pd.DataFrame({
    "income": np.random.lognormal(10.5, 0.6, 5000),
    "debt_ratio": np.random.beta(2, 5, 5000),
    "overdue_days": np.random.poisson(3, 5000)
})
prod_data = pd.DataFrame({
    "income": np.random.lognormal(10.8, 0.7, 5000),
    "debt_ratio": np.random.beta(2.5, 5, 5000),
    "overdue_days": np.random.poisson(5, 5000)
})

monitor_df = ks_drift_detection(ref_data, prod_data,
                                ["income", "debt_ratio", "overdue_days"],
                                alpha=0.05)
print(monitor_df)

输出示例：