Anomaly Transformer复现实战：从零搭建时间序列异常检测模型

共计 2034 个字符，预计需要花费 6 分钟才能阅读完成。

背景痛点：为什么需要关注时间序列异常检测？

工业设备监控、金融交易风控、IT 运维等领域都依赖时间序列异常检测。传统方法如统计阈值和孤立森林面临两个核心问题：

• 难以捕捉多元序列的复杂关联模式
• 对突发型异常（point anomaly）和持续型异常（collective anomaly）的区分能力弱

Anomaly Transformer 通过关联差异机制（Association Discrepancy）解决了这些问题，但原论文代码存在以下工程化问题：

1. 依赖已弃用的 TensorFlow 1.x API
2. 数据预处理与模型训练强耦合
3. 缺乏生产环境所需的推理优化

技术实现：PyTorch 实战三步走

1. 数据预处理标准化流程

def sliding_window(series: torch.Tensor, window: int, stride: int) -> torch.Tensor:
    """ 将单条时序数据转换为滑窗样本
    Args:
        series: [T, D] 原始时序
        window: 滑窗长度
        stride: 滑动步长
    Returns:
        [N, window, D] 样本集合
    """
    return series.unfold(0, window, stride).transpose(1, 2)

关键操作说明：

1. 使用 unfold 进行向量化滑窗（比 for 循环快 20 倍）
2. 通过 transpose 调整维度顺序符合 PyTorch 惯例

2. 关联差异机制核心实现

Association Discrepancy 包含两个关键组件：

• 先验关联（Prior-Association）：基于高斯先验的固定模式
• 序列关联（Series-Association）：通过 self-attention 学习到的动态模式

数学表达：
$$\mathcal{D}(\mathcal{A}^p, \mathcal{A}^s) = \sqrt{\frac{1}{L}\sum_{l=1}^L(\mathcal{A}^p_l – \mathcal{A}^s_l)^2}$$

代码实现：

class AnomalyAttention(nn.Module):
    def __init__(self, d_model: int):
        super().__init__()
        self.query = nn.Linear(d_model, d_model)
        self.key = nn.Linear(d_model, d_model)
        # 先验关联矩阵（不可训练）self.register_buffer('prior', self._gaussian_prior())

    def forward(self, x: Tensor) -> Tuple[Tensor, Tensor]:
        """x: [B, L, D]"""
        Q, K = self.query(x), self.key(x)  # [B, L, D]
        series_assoc = F.softmax(Q @ K.transpose(-2,-1), dim=-1)  # [B, L, L]
        discrepancy = (self.prior - series_assoc).pow(2).mean(-1).sqrt()  # [B, L]
        return series_assoc, discrepancy

3. PyTorch Lightning 训练架构

推荐使用 Lightning 的三大理由：

1. 自动处理 GPU/TPU 设备切换
2. 内置梯度裁剪和混合精度训练
3. 支持 TensorBoard 日志可视化

关键训练配置：

trainer:
  max_epochs: 100
  gradient_clip_val: 1.0
  precision: 16  # 混合精度训练

model:
  lr: 1e-4
  weight_decay: 1e-3
  prior_weight: 0.5  # 先验损失权重

生产环境优化技巧

推理延迟对比（Tesla T4）

ONNX 导出注意事项

1. 需固定输入序列长度
2. 禁用动态控制流
3. 验证输出误差小于 1e-5

torch.onnx.export(
    model, 
    dummy_input, 
    "model.onnx",
    input_names=["input"],
    dynamic_axes={"input": {0: "batch"}}  # 仅 batch 维度动态
)

避坑指南

CUDA 版本匹配

通过 conda install pytorch==1.12.1 cudatoolkit=11.3 -c pytorch 确保版本对应

长序列 OOM 解决方案

1. 梯度检查点技术

   model = torch.utils.checkpoint.checkpoint_sequential(model, chunks=4, input=x)

2. 减少 attention 头数（8→4）
3. 使用 batch_size=1 进行推理

先验权重调优

建议从 0.3 开始逐步增加，监控验证集 F1 分数变化

开放思考题

1. 如何设计在线学习机制应对数据分布漂移？
2. 关联差异机制能否应用于视频异常检测？
3. 先验知识应该完全固定还是允许微调？

完整 Colab 代码 包含 SMAP 数据集加载和可视化模块