多变量时序异常检测需要同时处理多个时间相关的变量，挖掘变量间的关联关系和时序变化规律，Transformer的自注意力机制能够有效捕捉长程依赖和变量间的交互信息，适合处理这类任务。下面我们将从数据准备到模型部署的完整流程展开说明。
核心原理概述
基于Transformer的多变量时序异常检测核心思路是，将多变量时序数据输入Transformer编码器，学习正常的时序模式，通过重构误差或者预测误差判断样本是否为异常。自注意力机制可以让模型关注不同时刻、不同变量之间的关系，比传统的循环神经网络更能捕捉复杂的时序特征。
数据预处理步骤
多变量时序数据的预处理是模型效果的基础，主要包含以下操作：
- 缺失值填充：采用线性插值或者均值填充的方式处理缺失的时间点数据
- 归一化：将每个变量的数值缩放到0到1的区间，避免不同量纲影响模型训练
- 滑动窗口构造：将连续的时序数据切分成固定长度的窗口样本，每个窗口作为一个输入样本
模型构建与训练
我们使用PyTorch构建基础的Transformer编码器模型，实现多变量时序的重构任务，正常样本的重构误差较低，异常样本的重构误差会明显偏高。
首先定义模型结构：
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from sklearn.preprocessing import MinMaxScaler
# 定义Transformer编码器模型
class TransformerAnomalyDetector(nn.Module):
def __init__(self, input_dim, d_model, nhead, num_encoder_layers, dim_feedforward, dropout=0.1):
super(TransformerAnomalyDetector, self).__init__()
self.input_projection = nn.Linear(input_dim, d_model)
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=dim_feedforward,
dropout=dropout,
batch_first=True
)
self.transformer_encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_encoder_layers)
self.output_projection = nn.Linear(d_model, input_dim)
def forward(self, x):
# x shape: (batch_size, seq_len, input_dim)
x = self.input_projection(x)
x = self.transformer_encoder(x)
x = self.output_projection(x)
return x
# 数据预处理函数
def preprocess_data(data, window_size=10):
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(data)
windows = []
for i in range(len(scaled_data) - window_size):
windows.append(scaled_data[i:i+window_size])
return np.array(windows), scaler
接下来是模型训练和异常判定的代码：
# 训练模型
def train_model(model, train_loader, epochs=20, lr=0.001):
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=lr)
model.train()
for epoch in range(epochs):
total_loss = 0
for batch in train_loader:
optimizer.zero_grad()
output = model(batch)
loss = criterion(output, batch)
loss.backward()
optimizer.step()
total_loss += loss.item()
print(f"Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.6f}")
# 异常检测函数
def detect_anomalies(model, data, scaler, window_size=10, threshold=None):
model.eval()
scaled_data = scaler.transform(data)
windows = []
for i in range(len(scaled_data) - window_size):
windows.append(scaled_data[i:i+window_size])
windows = torch.tensor(np.array(windows), dtype=torch.float32)
with torch.no_grad():
reconstructions = model(windows)
# 计算每个窗口的重构误差（均方误差）
errors = torch.mean((windows - reconstructions) ** 2, dim=(1,2)).numpy()
# 如果没有指定阈值，使用训练集误差的95分位数作为阈值
if threshold is None:
threshold = np.percentile(errors, 95)
anomalies = errors > threshold
# 将窗口级别的异常标记映射到原始时间点
anomaly_indices = []
for idx, is_anomaly in enumerate(anomalies):
if is_anomaly:
anomaly_indices.extend(range(idx, idx+window_size))
return anomaly_indices, errors, threshold
# 示例运行代码
if __name__ == "__main__":
# 构造示例多变量时序数据，3个变量，1000个时间点
np.random.seed(42)
normal_data = np.random.randn(1000, 3) * 0.5 + 1
# 插入10个异常点
anomaly_indices = np.random.choice(1000, 10, replace=False)
normal_data[anomaly_indices] += 5
# 预处理数据
window_size = 10
train_windows, scaler = preprocess_data(normal_data, window_size)
train_loader = torch.utils.data.DataLoader(torch.tensor(train_windows, dtype=torch.float32), batch_size=32, shuffle=True)
# 初始化模型
model = TransformerAnomalyDetector(
input_dim=3,
d_model=64,
nhead=4,
num_encoder_layers=2,
dim_feedforward=128
)
# 训练模型
train_model(model, train_loader, epochs=10)
# 检测异常
detected_anomalies, errors, threshold = detect_anomalies(model, normal_data, scaler, window_size)
print(f"检测到的异常时间点数量: {len(set(detected_anomalies))}")
print(f"异常阈值: {threshold:.6f}")
注意事项
实际应用中可以根据场景调整模型参数，比如增大d_model提升特征表达能力，增加num_encoder_layers捕捉更复杂的模式。如果数据量较大，可以适当减小batch_size避免显存溢出。异常阈值的选择需要结合业务场景调整，也可以通过验证集的标注数据优化阈值，提升检测准确率。
另外，如果时序数据存在明显的周期性，可以在输入模型前加入位置编码，帮助Transformer更好地捕捉时序的位置信息，进一步提升模型效果。
Transformer多变量时序异常检测Python时序数据处理修改时间：2026-07-19 08:45:39