USAD 算法深度解析与工业级时序异常检测实战

在智能运维（AIOps）和工业互联网领域，如何从海量的传感器数据中精准识别出设备故障、网络攻击或业务异常，一直是核心挑战。经典的自编码器（AE）虽然能学习重构，但在面对某些精巧的异常时往往表现得过于'宽容'，导致漏报。

2020 年，由 KTH 皇家理工学院提出的 USAD (UnSupervised Anomaly Detection for Multivariate Time Series) 算法横空出世。它巧妙地结合了**自编码器（AutoEncoder）的重构能力和生成对抗网络（GAN）**的对抗特性，成为了目前多维时序异常检测领域的标杆模型。

import torch import torch.nn as nn import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler # 1. 定义 USAD 模型结构 class Encoder(nn.Module): def __init__(self, in_size, latent_size): super().__init__() self.linear1 = nn.Linear(in_size, in_size // 2) self.linear2 = nn.Linear(in_size // 2, in_size // 4) self.linear3 = nn.Linear(in_size // 4, latent_size) self.relu = nn.ReLU() def forward(self, x): x = self.relu(self.linear1(x)) x = self.relu(self.linear2(x)) x = self.relu(self.linear3(x)) return x class Decoder(nn.Module): def __init__(self, latent_size, out_size): super().__init__() self.linear1 = nn.Linear(latent_size, out_size // 4) self.linear2 = nn.Linear(out_size // 4, out_size // 2) self.linear3 = nn.Linear(out_size // 2, out_size) self.relu = nn.ReLU() self.sigmoid = nn.Sigmoid() def forward(self, x): x = self.relu(self.linear1(x)) x = self.relu(self.linear2(x)) x = self.sigmoid(self.linear3(x)) return x class USAD(nn.Module): def __init__(self, w_size, z_size): super().__init__() self.encoder = Encoder(w_size, z_size) self.decoder1 = Decoder(z_size, w_size) self.decoder2 = Decoder(z_size, w_size) def training_step(self, batch, n): z = self.encoder(batch) w1 = self.decoder1(z) w2 = self.decoder2(z) w3 = self.decoder2(self.encoder(w1)) # Phase 1: 重构阶段 loss1 = 1/n * torch.mean(torch.abs(batch - w1)) + (1-1/n) * torch.mean(torch.abs(batch - w3)) # Phase 2: 对抗阶段 loss2 = 1/n * torch.mean(torch.abs(batch - w2)) - (1-1/n) * torch.mean(torch.abs(batch - w3)) return loss1, loss2 # 2. 数据处理：滑动窗口化 def create_windows(data, window_size): windows = [] for i in range(len(data) - window_size + 1): windows.append(data[i:i+window_size].reshape(-1)) return np.array(windows) # 3. 模拟实战流程 def run_usad_demo(): # 模拟数据生成：5 个维度的传感器数据 np.random.seed(42) normal_data = np.random.rand(1000, 5) test_data = np.random.rand(200, 5) test_data[50:60] += 2.0 # 注入异常 scaler = MinMaxScaler() normal_data = scaler.fit_transform(normal_data) test_data = scaler.transform(test_data) w_size = 10 # 窗口大小 train_windows = create_windows(normal_data, w_size) test_windows = create_windows(test_data, w_size) train_loader = torch.utils.data.DataLoader( torch.from_numpy(train_windows).float(), batch_size=64, shuffle=True) # 实例化模型 model = USAD(w_size * 5, z_size=20) opt1 = torch.optim.Adam(list(model.encoder.parameters()) + list(model.decoder1.parameters())) opt2 = torch.optim.Adam(list(model.encoder.parameters()) + list(model.decoder2.parameters())) # 训练循环 epochs = 20 for epoch in range(epochs): for batch in train_loader: loss1, loss2 = model.training_step(batch, epoch + 1) loss1.backward(retain_graph=True) opt1.step() opt1.zero_grad() loss2.backward() opt2.step() opt2.zero_grad() if epoch % 5 == 0: print(f"Epoch {epoch} complete") # 检测 model.eval() with torch.no_grad(): test_batch = torch.from_numpy(test_windows).float() z = model.encoder(test_batch) w1 = model.decoder1(z) w2 = model.decoder2(model.encoder(w1)) # 计算异常得分 scores = 0.5 * torch.mean(torch.abs(test_batch - w1), axis=1) + \ 0.5 * torch.mean(torch.abs(test_batch - w2), axis=1) return scores if __name__ == "__main__": scores = run_usad_demo() print("Top 5 anomaly scores:", scores[:5].numpy())

USAD 算法深度解析与工业级时序异常检测实战