异常检测算法初识：从 3-Sigma 统计学到动态时间规整 DTW

异常检测算法初识：从 3-Sigma 统计学到动态时间规整 DTW

在智能运维（AIOps）的时序监控中，我们经常遇到一类非常特殊的“形态异常（Shape Anomaly）”：

• 某核心业务在每天早上 09:00 会准时迎来早高峰流量爬坡，耗时从 40ms 上涨至 90ms；
• 但今天早高峰由于外部网络波动，流量爬坡推迟到了 09:20 才发生。

如果使用传统的 3-Sigma 或 欧氏距离（Euclidean Distance） 逐点对齐比对，算法会在 09:0009:20 之间疯狂报警（因为此时实际值远低于历史期望均值），并在 09:2009:40 再次疯狂报警（实际值远高于历史同期）。但从业务全局来看，这只是一个在时间轴上发生了轻微拉伸或相位延迟的正常业务周期，根本不是系统故障。

传统的逐点对齐算法无法处理时间轴上的局部形变与速度漂移。引入源自语音识别领域的 动态时间规整算法（Dynamic Time Warping, DTW），是解决时序形态相似度计算与自适应异常检测的破局利器。

欧氏距离 vs 动态时间规整（DTW）

【欧氏距离 (Euclidean Distance – 一对一硬对齐)】
序列 A (今日时序): o───o───o───────o───o (波峰在第 4 点)
│ │ │ │ │
序列 B (历史基线): o───o───────o───o───o (波峰在第 3 点)
==> 缺陷: 强行在相同时间戳求差值平方和，时间轴轻微平移导致误差急剧放大

【DTW 动态时间规整 (一对多弹性扭曲对齐)】
序列 A (今日时序): o───o───o───────o───o
\\ \\ \\ / /
序列 B (历史基线): o───o───o───o───o
==> 优势: 允许在时间轴上“压缩或拉伸”，找到两个时序之间距离最小的最优规整路径

DTW 算法的核心数学原理

给定两个时序序列 $X = (x_1, x_2, \\dots, x_n)$ 和 $Y = (y_1, y_2, \\dots, y_m)$：

$$\\gamma(i, j) = D(i, j) + \\min \\Big( \\gamma(i-1, j), , \\gamma(i, j-1), , \\gamma(i-1, j-1) \\Big)$$

最终 $\\gamma(n, m)$ 即为两个时序在最优时间扭曲下的最小全局形态距离。

核心实现：基于 Python 的快速 DTW 形态异常检测器

标准 DTW 的时间复杂度为 $O(n \\cdot m)$。在生产时序监控中，通常引入 Sakoe-Chiba 约束窗口（Warping Window Band） 将计算复杂度优化至接近线性 $O(n \\cdot w)$：

import numpy as np
from dataclasses import dataclass
from typing import List, Tuple

@dataclass
class DtwAnomalyResult:
is_anomaly: bool
dtw_distance: float
threshold: float
warping_path_length: int

class FastDtwDetector:
def __init__(self, window_size: int = 5, distance_threshold: float = 15.0):
"""
window_size: Sakoe-Chiba 带宽限制 (允许的最大时间漂移步数)
distance_threshold: 形态偏离报警阈值
"""
self.w = window_size
self.threshold = distance_threshold

def compute_dtw_distance(self, s1: np.ndarray, s2: np.ndarray) -> float:
n, m = len(s1), len(s2)
w = max(self.w, abs(n – m))

# 初始化累积代价矩阵
dtw_matrix = np.full((n + 1, m + 1), np.inf)
dtw_matrix[0, 0] = 0.0

for i in range(1, n + 1):
# 仅在对角线附近带宽内计算
for j in range(max(1, i – w), min(m + 1, i + w + 1)):
cost = (s1[i – 1] – s2[j – 1]) ** 2
dtw_matrix[i, j] = cost + min(
dtw_matrix[i – 1, j], # 插入 (时间拉伸)
dtw_matrix[i, j – 1], # 删除 (时间压缩)
dtw_matrix[i – 1, j – 1] # 匹配
)

return float(np.sqrt(dtw_matrix[n, m]))

def evaluate_shape_anomaly(self, current_series: List[float], baseline_series: List[float]) -> DtwAnomalyResult:
s1 = np.array(current_series, dtype=float)
s2 = np.array(baseline_series, dtype=float)

# Z-Score 标准化，消除整体量纲与绝对值偏移，专注于“形态曲线变化”
s1_norm = (s1 – np.mean(s1)) / (np.std(s1) + 1e-6)
s2_norm = (s2 – np.mean(s2)) / (np.std(s2) + 1e-6)

dist = self.compute_dtw_distance(s1_norm, s2_norm)
is_anomaly = dist > self.threshold

return DtwAnomalyResult(
is_anomaly=is_anomaly,
dtw_distance=round(dist, 2),
threshold=self.threshold,
warping_path_length=len(s1)
)

# 测试验证
detector = FastDtwDetector(window_size=4, distance_threshold=8.0)

# 基准序列：标准爬坡波形
baseline = [10, 12, 15, 45, 80, 95, 100, 98, 85, 40, 20, 10]

# 测试用例 1：时间滞后 2 个周期的正常爬坡（纯时间轴拉伸，形态一致）
lagged_normal = [10, 10, 10, 13, 16, 46, 82, 96, 101, 97, 83, 38]
res1 = detector.evaluate_shape_anomaly(lagged_normal, baseline)
print(f"时移正常波形检测: 异常={res1.is_anomaly}, DTW距离={res1.dtw_distance} (欧氏距离极高但DTW识别为正常)")

# 测试用例 2：突发暴跌畸形波（中途断崖式下跌，形态严重畸变）
abnormal_shape = [10, 12, 15, 45, 12, 10, 10, 15, 85, 40, 20, 10]
res2 = detector.evaluate_shape_anomaly(abnormal_shape, baseline)
print(f"畸形波形检测: 异常={res2.is_anomaly}, DTW距离={res2.dtw_distance} (成功触发报警)")

在 AIOps 生产环境中的应用场景

双模结合，让智能运维系统兼具敏锐的嗅觉与大局观。