4、概念漂移检测原理：DDM、ADWIN、Page-Hinkley 方法

概念漂移（Concept Drift）是 MRM（Model Risk Management）中导致模型性能衰退的核心原因之一。当数据分布随时间发生变化，模型在训练集上习得的决策边界不再适用于当前环境，预测误差便会显著上升。本章深入剖析三种经典的概念漂移检测算法：DDM（Drift Detection Method）、ADWIN（Adaptive Windowing）和 Page-Hinkley 方法，帮助运维专家在监控系统中精准定位漂移发生的时间点。

4.1 三种检测方法的核心思想对比

在深入算法细节之前，先通过下表建立对三种方法的宏观认知。它们都基于在线学习（Online Learning）框架，逐样本或逐批次更新检测状态。

4.2 DDM（Drift Detection Method）

DDM 由 Gama 等人于 2004 年提出，是最经典的概念漂移检测算法之一。它假设模型在稳定环境下，预测错误率服从二项分布，并随着样本量增加逐渐收敛。当错误率显著偏离历史均值时，触发漂移警告。

4.2.1 算法原理与阈值

DDM 维护两个关键统计量：当前平均错误率 \( p_i \) 和 标准差 \( s_i = \sqrt{p_i (1-p_i) / i} \)。随着样本数 \( i \) 增加，记录历史最小错误率 \( p_{min} \) 及其标准差 \( s_{min} \)。

• 警告阶段（Warning）： 当 \( p_i + s_i \geq p_{min} + 2 \cdot s_{min} \) 时，进入警告区，开始缓存后续样本，准备可能的模型更新。
• 漂移阶段（Drift）： 当 \( p_i + s_i \geq p_{min} + 3 \cdot s_{min} \) 时，确认发生概念漂移，触发模型重训练或回滚。
• 恢复机制： 漂移触发后，重置统计量，从零开始重新监控。

# DDM 伪代码示例
class DDM:
    def __init__(self, warning_ratio=2.0, drift_ratio=3.0):
        self.n = 0
        self.error_count = 0
        self.p_min = float('inf')
        self.s_min = float('inf')
        self.warning_ratio = warning_ratio
        self.drift_ratio = drift_ratio
        self.state = 'normal'  # normal, warning, drift

    def add_element(self, is_error: bool):
        self.n += 1
        if is_error:
            self.error_count += 1
        p = self.error_count / self.n
        s = (p * (1 - p) / self.n) ** 0.5

        if p + s < self.p_min + self.s_min:
            self.p_min = p
            self.s_min = s

        if p + s > self.p_min + self.drift_ratio * self.s_min:
            self.state = 'drift'
            self.reset()
        elif p + s > self.p_min + self.warning_ratio * self.s_min:
            self.state = 'warning'
        else:
            self.state = 'normal'
        return self.state

4.3 ADWIN（Adaptive Windowing）

ADWIN 由 Bifet 和 Gavaldà 于 2007 年提出，核心思想是维护一个可变长度的滑动窗口，当窗口内数据分布发生显著变化时，自动丢弃窗口前部的旧数据。ADWIN 不需要预设窗口大小，完全由数据驱动。

4.3.1 自适应窗口裁剪机制

ADWIN 维护一个窗口 \( W \)，将其划分为两个子窗口 \( W_0 \)（旧数据）和 \( W_1 \)（新数据）。算法持续检验两个子窗口的均值是否存在统计显著差异。如果差异超过阈值 \( \epsilon_{cut} \)，则裁剪掉 \( W_0 \)，只保留 \( W_1 \) 作为新的窗口。

裁剪阈值由 Hoeffding 不等式推导得出：

• \( \epsilon_{cut} = \sqrt{\frac{1}{2 \cdot m} \cdot \ln \frac{4 \cdot n}{\delta}} \)，其中 \( m = \frac{1}{1/|W_0| + 1/|W_1|} \)，\( n = |W| \)，\( \delta \) 为置信度参数（通常取 0.05 或 0.01）。
• 当 \( |\mu_{W_0} - \mu_{W_1}| > \epsilon_{cut} \) 时，判定发生漂移，丢弃 \( W_0 \)。

# ADWIN 核心逻辑伪代码
class ADWIN:
    def __init__(self, delta=0.05):
        self.window = []          # 存储 (value, timestamp)
        self.total = 0.0
        self.width = 0
        self.delta = delta

    def add_element(self, value: float):
        self.window.append((value, self.width))
        self.total += value
        self.width += 1
        # 尝试裁剪窗口
        while len(self.window) > 1:
            # 遍历所有可能的分割点
            for cut_idx in range(1, len(self.window)):
                w0 = self.window[:cut_idx]
                w1 = self.window[cut_idx:]
                mu0 = sum(v for v,_ in w0) / len(w0)
                mu1 = sum(v for v,_ in w1) / len(w1)
                m = 1.0 / (1.0/len(w0) + 1.0/len(w1))
                eps = ( (1/(2*m)) * math.log(4*self.width/self.delta) ) ** 0.5
                if abs(mu0 - mu1) > eps:
                    # 漂移：丢弃 w0
                    self.window = w1
                    self.total = sum(v for v,_ in w1)
                    self.width = len(w1)
                    break
            else:
                break  # 没有发生裁剪
        return self.get_mean()

    def get_mean(self):
        return self.total / self.width if self.width > 0 else 0.0

4.4 Page-Hinkley 方法

Page-Hinkley 方法（PH 检验）源自序贯分析（Sequential Analysis），最初用于信号处理中的变化点检测。它通过累积观测值与参考均值的偏差，当累积和超过阈值时判定发生漂移。PH 方法计算极轻量，适合对延迟敏感的实时监控场景。

4.4.1 累积偏差与阈值判定

PH 方法维护两个累积统计量：

• 累积和 \( U_T \)： \( U_T = \sum_{t=1}^{T} (x_t - \bar{x}_T - \delta) \)，其中 \( \bar{x}_T \) 为到当前时刻的均值，\( \delta \) 为容忍偏移量（通常取 0.01 或 0.001）。
• 最小累积和 \( m_T \)： \( m_T = \min_{1 \leq i \leq T} U_i \)。
• PH 统计量： \( PH_T = U_T - m_T \)。
• 漂移判定： 当 \( PH_T > \lambda \) 时触发漂移，其中 \( \lambda \) 为预设阈值（通常取 50~200）。

# Page-Hinkley 方法实现
class PageHinkley:
    def __init__(self, threshold=100.0, delta=0.005, alpha=0.9999):
        self.threshold = threshold   # 漂移阈值 λ
        self.delta = delta           # 容忍偏移量
        self.alpha = alpha           # 遗忘因子（可选，用于非平稳环境）
        self.n = 0
        self.sum = 0.0               # 累积和
        self.mean = 0.0              # 运行均值
        self.U = 0.0                 # 累积偏差
        self.m = 0.0                 # 最小累积偏差
        self.PH = 0.0                # PH 统计量

    def add_element(self, x: float):
        self.n += 1
        # 更新运行均值（带遗忘因子）
        self.mean = self.alpha * self.mean + (1 - self.alpha) * x
        # 累积偏差
        self.U += (x - self.mean - self.delta)
        # 更新最小值
        if self.U < self.m:
            self.m = self.U
        # 计算 PH 统计量
        self.PH = self.U - self.m
        # 判定漂移
        if self.PH > self.threshold:
            self.reset()
            return 'drift'
        return 'normal'

    def reset(self):
        self.n = 0
        self.sum = 0.0
        self.mean = 0.0
        self.U = 0.0
        self.m = 0.0
        self.PH = 0.0

4.5 三种方法的工程选型矩阵

在实际 MRM 平台中，通常不会只依赖单一检测方法，而是组合使用。下表从工程部署角度给出选型建议：

三种方法各有优劣，没有银弹。DDM 简单直观，适合分类场景的快速预警；ADWIN 自适应能力强，是处理复杂漂移模式的首选；Page-Hinkley 计算极轻，适合资源受限的部署环境。在 MRM 实践中，建议根据模型类型、监控指标性质和业务容忍度，选择最匹配的方法或组合方案。