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

3.2 对抗损失 (Adversarial Loss)

在 GAN 中，损失函数是零和博弈。但在 USAD 中，对抗损失被转化为重构误差的对比。Decoder2 尝试最小化重构'被 Decoder1 重构后的数据'的误差，而 Encoder 尝试最大化这个误差。

四、实战演练：构建基于 USAD 的多维指标监测系统

本文将使用 Python 和 PyTorch 实现一个可直接运行的 USAD 模型，用于检测模拟的工业传感器异常。

4.1 环境准备

建议使用 Python 3.8+ 环境。

pip install torch numpy pandas scikit-learn matplotlib

4.2 核心模型实现 (code.py)

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())

4.3 预期执行结果

执行该代码后，你会看到模型在 Epoch 15-20 左右趋于稳定。由于我们在 test_data[50:60] 处注入了巨大的偏移值，反映在 scores 结果中，索引 50 附近的数值会远高于其他区域（通常高出 10-50 倍），此时通过设置阈值（如均值的 3 倍标准差）即可实现自动报警。

五、深度进阶：如何在生产环境中落地 USAD？

在生产环境部署时建议考虑以下几点：

5.1 阈值自动选择

不要使用固定阈值。推荐使用 POT (Peak Over Threshold) 算法。它基于极值理论，可以根据数据分布自动计算出一个动态的、科学的异常边界。

5.2 模型剪枝与加速

USAD 的全连接层虽然简单，但在处理 100 维以上的指标时速度会变慢。可以考虑使用 TensorRT 进行模型量化，或者将 PyTorch 模型转化为 ONNX 格式，从而在 CPU 上也能达到毫秒级的推理速度。

5.3 增量学习 (Incremental Learning)

工业环境的'正常'定义是会随季节或生产计划改变的。建议每隔一周使用最近的正常数据对 USAD 进行 Fine-tuning，避免模型因环境漂移导致的误报。

六、总结

USAD 算法通过简洁而巧妙的对抗自编码器架构，解决了多维时序异常检测中'重构过强'的痛点。它既具备深度学习的非线性拟合能力，又通过对抗机制增强了对微小异常的敏感度。

无论你是要监控 Kubernetes 集群的性能指标，还是检测风力发电机的传感器数据，USAD 都是一个非常扎实且易于落地的起点。