4. 统计检验方法（一）：KS检验（Kolmogorov-Smirnov Test）

KS检验（Kolmogorov-Smirnov Test）是特征漂移监控中最经典的非参数统计检验方法之一。它通过比较两个样本的累积分布函数（CDF）之间的最大垂直距离，来判断它们是否来自同一分布。在MLOps实践中，KS检验常用于检测数值型特征的分布变化。

4.1 原理：最大垂直距离度量分布差异

KS检验的核心思想非常直观：如果两个样本来自同一分布，那么它们的经验累积分布函数（ECDF）应该非常接近；反之，如果分布发生漂移，ECDF之间会出现明显的分离。KS统计量定义为两个ECDF之间的最大绝对差：

KS检验的假设体系：

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

当计算出的D值超过某个临界值（或对应的p值小于显著性水平α）时，拒绝原假设，判定特征发生漂移。

4.2 适用场景与局限性

维度	说明
适用数据类型	连续数值型特征（如年龄、收入、温度、响应时间）
样本量要求	两样本量均 ≥ 50 时效果较好；小样本时检验力不足
分布假设	无参数假设，不要求数据服从正态分布
敏感性	对分布中心位置、形状、离散度的变化均敏感
主要局限	对尾部差异不敏感；不适用于离散/分类特征；大样本时易过度敏感

4.3 Python实现：scipy.stats.ks_2samp

Python中通过 scipy.stats.ks_2samp 函数即可完成双样本KS检验。以下是一个完整的监控实现示例：

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

def ks_drift_detection(reference, current, alpha=0.05):
    """
    使用KS检验检测特征漂移
    
    参数:
        reference: array-like, 参考数据（训练集）
        current:   array-like, 当前数据（生产环境）
        alpha:     float, 显著性水平，默认0.05
    
    返回:
        dict: 包含统计量、p值、是否漂移、建议阈值
    """
    # 执行双样本KS检验
    statistic, p_value = stats.ks_2samp(reference, current)
    
    # 判断是否拒绝原假设
    drift_detected = p_value < alpha
    
    # 计算建议的漂移阈值（基于样本量）
    n1, n2 = len(reference), len(current)
    critical_value = np.sqrt(-0.5 * np.log(alpha / 2)) * np.sqrt((n1 + n2) / (n1 * n2))
    
    return {
        'ks_statistic': round(statistic, 4),
        'p_value': round(p_value, 6),
        'drift_detected': drift_detected,
        'critical_value': round(critical_value, 4),
        'alpha': alpha
    }

# 示例：模拟正常数据和漂移数据
np.random.seed(42)
reference_data = np.random.normal(loc=50, scale=10, size=1000)
current_normal = np.random.normal(loc=50, scale=10, size=1000)   # 无漂移
current_drift  = np.random.normal(loc=55, scale=12, size=1000)   # 均值+5，标准差+2

# 检测无漂移场景
result_normal = ks_drift_detection(reference_data, current_normal)
print("=== 无漂移场景 ===")
print(f"KS统计量: {result_normal['ks_statistic']}")
print(f"p值: {result_normal['p_value']}")
print(f"是否漂移: {result_normal['drift_detected']}")

# 检测有漂移场景
result_drift = ks_drift_detection(reference_data, current_drift)
print("\n=== 存在漂移场景 ===")
print(f"KS统计量: {result_drift['ks_statistic']}")
print(f"p值: {result_drift['p_value']}")
print(f"是否漂移: {result_drift['drift_detected']}")

4.4 结果解读与阈值设定

KS检验的输出需要结合业务上下文进行解读：

• KS统计量 D：取值范围 [0, 1]，值越大表示分布差异越大。通常 D > 0.1 时值得关注，D > 0.2 时表示显著漂移。
• p值：当 p < α（通常 0.05）时拒绝原假设，判定存在统计显著的分布差异。
• 临界值：基于样本量自动计算，当 D > 临界值时判定漂移。

4.5 多特征批量监控实现

在实际MLOps系统中，通常需要同时对数十甚至数百个特征进行漂移监控。以下是一个批量监控的实现模式：

def batch_ks_monitor(reference_df, current_df, alpha=0.05):
    """
    批量监控多个数值特征的KS漂移
    
    参数:
        reference_df: DataFrame, 参考数据集
        current_df:   DataFrame, 当前数据集
        alpha:        float, 显著性水平
    
    返回:
        DataFrame: 每个特征的漂移检测结果
    """
    results = []
    
    for col in reference_df.columns:
        # 跳过非数值列
        if not np.issubdtype(reference_df[col].dtype, np.number):
            continue
        
        ref_data = reference_df[col].dropna().values
        cur_data = current_df[col].dropna().values
        
        # 执行KS检验
        stat, p_val = stats.ks_2samp(ref_data, cur_data)
        
        results.append({
            'feature': col,
            'ks_statistic': round(stat, 4),
            'p_value': round(p_val, 6),
            'drift': p_val < alpha,
            'sample_size_ref': len(ref_data),
            'sample_size_cur': len(cur_data)
        })
    
    result_df = pd.DataFrame(results)
    result_df = result_df.sort_values('ks_statistic', ascending=False)
    
    return result_df

# 使用示例
# drift_report = batch_ks_monitor(train_features, production_features)
# drift_report[drift_report['drift'] == True]  # 查看所有漂移特征

4.6 关键注意事项

• 样本量匹配：KS检验对样本量差异敏感，建议参考样本和当前样本量尽量接近，避免一方过小导致检验力不足。
• 多重比较校正：当同时监控大量特征时，建议对p值进行Bonferroni或FDR校正，控制全局假阳性率。
• 业务阈值优先：在工业实践中，建议优先使用基于业务经验的KS阈值（如D > 0.15），而非仅依赖统计显著性。
• 可视化辅助：每次检测后应同时绘制ECDF对比图，帮助团队直观理解漂移的具体位置和幅度。