2. 基于自编码器的异常检测模型构建
任务说明
本案例面向旋转机械振动监测场景，构建一个基于自编码器的异常检测模型。输入为正常工况下的振动特征向量（如均方根值、峰值因子、峭度等），模型通过无监督学习压缩并重建正常数据分布；当输入包含异常样本（如轴承早期故障）时，重建误差显著增大，据此判定异常。最终交付可运行的Python训练脚本与推理函数，并给出判定阈值设定方法。
数据准备
由于本案例聚焦模型构建方法，使用公开可用且可复现的合成数据模拟正常振动特征。生成500个正常样本（特征维度为6），其中前3维服从均值为0.5、标准差为0.1的正态分布，后3维服从均值为0.3、标准差为0.08的正态分布；另生成50个异常样本，其部分特征偏移1.5～2个标准差。数据生成使用固定随机种子以保证可复现性。
模型架构与实现
自编码器包含编码器（6→4→2）与解码器（2→4→6），中间层使用ReLU激活，输出层使用Sigmoid将重建值约束到[0,1]范围。训练时以均方误差（MSE）作为重建损失，优化器采用Adam，学习率0.001，批量大小32，训练轮数100。训练完成后，计算训练集所有样本重建误差的均值与标准差，以“均值+3×标准差”作为异常判定阈值。
import numpy as np
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
# ---------- 1. 生成可复现的合成数据 ----------
np.random.seed(42)
torch.manual_seed(42)
n_normal = 500
n_anomaly = 50
feature_dim = 6
# 正常样本：两个高斯簇叠加
normal_data = np.concatenate([
np.random.normal(0.5, 0.1, (n_normal, 3)),
np.random.normal(0.3, 0.08, (n_normal, 3))
], axis=1)
# 异常样本：部分特征偏移
anomaly_data = np.concatenate([
np.random.normal(0.5, 0.1, (n_anomaly, 2)),
np.random.normal(0.9, 0.15, (n_anomaly, 2)), # 第3、4维偏移
np.random.normal(0.3, 0.08, (n_anomaly, 2))
], axis=1)
# 归一化到[0,1]（用正常样本的min/max）
all_normal = normal_data
min_vals = all_normal.min(axis=0)
max_vals = all_normal.max(axis=0)
def normalize(X):
return (X - min_vals) / (max_vals - min_vals + 1e-8)
X_train = normalize(normal_data).astype(np.float32)
X_test_normal = normalize(normal_data[:100]).astype(np.float32) # 留出100个正常样本测试
X_test_anomaly = normalize(anomaly_data).astype(np.float32)
# ---------- 2. 定义自编码器 ----------
class Autoencoder(nn.Module):
def __init__(self, input_dim=6):
super().__init__()
self.encoder = nn.Sequential(
nn.Linear(input_dim, 4),
nn.ReLU(),
nn.Linear(4, 2),
nn.ReLU()
)
self.decoder = nn.Sequential(
nn.Linear(2, 4),
nn.ReLU(),
nn.Linear(4, input_dim),
nn.Sigmoid()
)
def forward(self, x):
return self.decoder(self.encoder(x))
# ---------- 3. 训练 ----------
model = Autoencoder()
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
dataset = TensorDataset(torch.from_numpy(X_train))
loader = DataLoader(dataset, batch_size=32, shuffle=True)
epochs = 100
for epoch in range(epochs):
for batch in loader:
x = batch[0]
recon = model(x)
loss = criterion(recon, x)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# ---------- 4. 计算阈值 ----------
model.eval()
with torch.no_grad():
train_recon = model(torch.from_numpy(X_train))
train_errors = ((train_recon - torch.from_numpy(X_train)) ** 2).mean(dim=1).numpy()
threshold = train_errors.mean() + 3 * train_errors.std()
print(f"训练集重建误差均值: {train_errors.mean():.4f}")
print(f"异常判定阈值: {threshold:.4f}")
# ---------- 5. 推理与判定 ----------
def detect_anomaly(feature_vector):
"""输入归一化后的特征向量，返回(是否异常, 重建误差)"""
model.eval()
x = torch.from_numpy(np.array(feature_vector, dtype=np.float32).reshape(1, -1))
with torch.no_grad():
recon = model(x)
error = ((recon - x) ** 2).mean().item()
return error > threshold, error
# ---------- 6. 验证 ----------
with torch.no_grad():
normal_errors = ((model(torch.from_numpy(X_test_normal)) - torch.from_numpy(X_test_normal)) ** 2).mean(dim=1).numpy()
anomaly_errors = ((model(torch.from_numpy(X_test_anomaly)) - torch.from_numpy(X_test_anomaly)) ** 2).mean(dim=1).numpy()
normal_detect_rate = (normal_errors < threshold).mean()
anomaly_detect_rate = (anomaly_errors > threshold).mean()
print(f"正常样本判定正确率: {normal_detect_rate:.2%}")
print(f"异常样本检出率: {anomaly_detect_rate:.2%}")
# 单样本推理示例
sample = X_test_anomaly[0]
is_anomaly, err = detect_anomaly(sample)
print(f"单样本推理 - 是否异常: {is_anomaly}, 重建误差: {err:.4f}")落地与运行方式
将上述代码保存为ae_anomaly_detection.py，在已安装PyTorch（≥1.9）与NumPy的Python 3.8+环境中运行：
pip install torch numpy
python ae_anomaly_detection.py验证方法与判定标准
运行脚本后观察终端输出：
- 训练集重建误差均值与阈值正常打印，无NaN或异常数值；
- 正常样本判定正确率应高于95%（因正常测试样本与训练同分布）；
- 异常样本检出率应高于90%（因异常特征偏移明显）；
- 单样本推理输出布尔值与误差数值，与代码逻辑一致。
满足以上条件即判定模型构建成功。若检出率偏低，可检查数据归一化是否使用训练集统计量，或适当增加训练轮数。
模型结构与判定流程
图中展示了自编码器的完整数据流：输入6维特征经编码器压缩至2维瓶颈表示，再经解码器重建为6维输出；计算输入与输出的MSE作为异常分数，与训练集统计阈值比较后输出判定结果。该流程可直接嵌入PHM系统的在线监测模块，对实时振动特征进行逐样本异常筛查。