航天器遥测数据异常检测实战：基于 PyTorch Geometric 实现 MAG 模型，窗口大小 50 步长 1

航天器遥测数据异常检测实战：基于 PyTorch Geometric 实现 MAG 模型

航天器在轨运行期间产生的遥测数据如同精密仪器的生命体征，包含着反映系统健康状态的丰富信息。这些数据通常呈现为高维、非线性且具有复杂时间依赖性的多变量时间序列，传统检测方法往往难以捕捉其深层特征。本文将手把手带您实现基于最大信息系数注意力图网络（MAG）的异常检测系统，使用 PyTorch Geometric 构建图神经网络，针对窗口大小为50、步长为1的滑动窗口数据进行实战建模。

1. 环境配置与数据准备

1.1 基础环境搭建

确保使用 CUDA 11.6 和 PyTorch 1.9.1 版本以获得最佳兼容性。以下是完整的依赖安装命令：

conda create -n mag_env python=3.8 conda activate mag_env pip install torch==1.9.1+cu116 torchvision==0.10.1+cu116 -f https://download.pytorch.org/whl/torch_stable.html pip install torch-geometric==1.7.2 torch-scatter torch-sparse -f https://data.pyg.org/whl/torch-1.9.1+cu116.html pip install minepy pandas scikit-learn

1.2 NASA 数据集预处理

以 SMAP/MSL 数据集为例，原始数据需要经过以下处理流程：

import pandas as pd import numpy as np def load_and_preprocess(data_path): # 读取原始遥测数据 raw_data = pd.read_csv(data_path, parse_dates=['timestamp']) # 标准化处理（保留状态变量的二进制特性） cont_vars = [col for col in raw_data.columns if col not in ['timestamp', 'anomaly', 'status_']] stat_vars = [col for col in raw_data.columns if col.startswith('status_')] # 对连续变量进行标准化 data_mean = raw_data[cont_vars].mean() data_std = raw_data[cont_vars].std() raw_data[cont_vars] = (raw_data[cont_vars] - data_mean) / data_std # 滑动窗口生成（窗口50，步长1） window_size = 50 stride = 1 sequences = [] labels = [] for i in range(0, len(raw_data) - window_size + 1, stride): window = raw_data.iloc[i:i+window_size] sequences.append(window[cont_vars + stat_vars].values) labels.append(window['anomaly'].max()) # 窗口内任一时刻异常则标记为异常 return np.array(sequences), np.array(labels), cont_vars, stat_vars

关键参数说明：

• 窗口设计：50个时间步的窗口可平衡特征捕获与实时性需求
• 变量处理：连续变量标准化，状态变量保持原始二进制形式
• 标签策略：采用"窗口内任一异常即整体异常"的严格标准

2. 图结构构建与特征工程

2.1 最大信息系数（MIC）计算

使用 minepy 计算变量间的非线性相关性：

from minepy import MINE def compute_mic_matrix(data, var_names): n_vars = len(var_names) mic_matrix = np.zeros((n_vars, n_vars)) mine = MINE(alpha=0.6, c=15) for i in range(n_vars): for j in range(i, n_vars): mine.compute_score(data[:, i], data[:, j]) mic_matrix[i, j] = mine.mic() mic_matrix[j, i] = mic_matrix[i, j] return mic_matrix

2.2 动态图结构生成

每个时间窗口构建一个动态图，节点特征包含：

边权重计算融合MIC和注意力机制：

import torch import torch.nn as nn class EdgeConstructor(nn.Module): def __init__(self, embed_dim): super().__init__() self.embed_dim = embed_dim self.attention = nn.Sequential( nn.Linear(2 * embed_dim, embed_dim), nn.ReLU(), nn.Linear(embed_dim, 1) ) def forward(self, node_embeddings, mic_matrix): # 计算注意力系数 n_nodes = node_embeddings.size(0) edge_indices = [] edge_weights = [] for i in range(n_nodes): for j in range(n_nodes): if mic_matrix[i,j] > 0.3: # MIC阈值过滤 attn_input = torch.cat([node_embeddings[i], node_embeddings[j]], dim=-1) alpha_ij = torch.sigmoid(self.attention(attn_input)) e_ij = mic_matrix[i,j] * alpha_ij edge_indices.append([i,j]) edge_weights.append(e_ij) return torch.tensor(edge_indices).t().contiguous(), torch.stack(edge_weights).squeeze()

3. MAG 模型实现

3.1 模型架构设计

完整 MAG 模型包含以下核心组件：

from torch_geometric.nn import GATConv import torch.nn.functional as F class MAGModel(nn.Module): def __init__(self, num_vars, cont_dim, stat_dim, embed_dim=128): super().__init__() # 变量嵌入层 self.var_embedding = nn.Embedding(num_vars, embed_dim) # 时间特征提取 self.lstm = nn.LSTM(cont_dim + stat_dim, embed_dim // 2, bidirectional=True, batch_first=True) # 图注意力网络 self.gat1 = GATConv(embed_dim * 2, embed_dim, heads=3) self.gat2 = GATConv(embed_dim * 3, embed_dim) # 预测头 self.cont_head = nn.Linear(embed_dim, cont_dim) self.stat_head = nn.Linear(embed_dim, stat_dim) def forward(self, x, edge_index, edge_weight): # x: (batch_size, window_size, num_features) batch_size, window_size, num_features = x.shape num_vars = num_features # 生成节点特征 var_ids = torch.arange(num_vars).repeat(batch_size, 1).to(x.device) static_embeds = self.var_embedding(var_ids) # (batch, num_vars, embed_dim) # 提取时间特征 temporal_feats, _ = self.lstm(x) # (batch, window, embed_dim) temporal_feats = temporal_feats[:, -1] # 取最后时间步 # 合并特征 node_feats = torch.cat([static_embeds, temporal_feats.unsqueeze(1).repeat(1, num_vars, 1)], dim=-1) # 图神经网络处理 h = F.relu(self.gat1(node_feats, edge_index, edge_weight)) h = self.gat2(h, edge_index, edge_weight) # 多任务输出 cont_pred = self.cont_head(h) stat_pred = torch.sigmoid(self.stat_head(h)) return cont_pred, stat_pred

3.2 混合损失函数

针对连续变量和状态变量的不同特性设计损失函数：

def hybrid_loss(cont_pred, cont_true, stat_pred, stat_true, lambda_reg=0.01): # 连续变量使用MSE mse_loss = F.mse_loss(cont_pred, cont_true) # 状态变量使用BCE bce_loss = F.binary_cross_entropy(stat_pred, stat_true) # 图结构正则化 l2_reg = torch.tensor(0.).to(cont_pred.device) for param in model.parameters(): l2_reg += torch.norm(param) total_loss = mse_loss + bce_loss + lambda_reg * l2_reg return total_loss

4. 训练与异常检测

4.1 自适应阈值计算

采用基于中位数和四分位距的稳健阈值：

def compute_adaptive_threshold(train_errors): """ train_errors: 训练集上的预测误差数组 返回: (median, iqr, threshold) """ median = np.median(train_errors) q75, q25 = np.percentile(train_errors, [75, 25]) iqr = q75 - q25 threshold = median + 3 * iqr # 3IQR准则 return threshold

4.2 完整训练流程

from torch_geometric.data import Data from sklearn.model_selection import train_test_split # 数据准备 sequences, labels, cont_vars, stat_vars = load_and_preprocess('smap_data.csv') mic_matrix = compute_mic_matrix(sequences.reshape(-1, len(cont_vars)+len(stat_vars)), cont_vars + stat_vars) # 数据集划分 X_train, X_val, y_train, y_val = train_test_split(sequences, labels, test_size=0.2, random_state=42) # 转换为PyG数据格式 train_dataset = [] for seq in X_train: # 每个样本构建一个图 edge_index, edge_weight = edge_constructor(seq) data = Data(x=torch.FloatTensor(seq[-1]), # 最后时间步作为节点特征 edge_index=edge_index, edge_attr=edge_weight, y=torch.FloatTensor([seq[-1]])) train_dataset.append(data) # 训练循环 model = MAGModel(len(cont_vars)+len(stat_vars), len(cont_vars), len(stat_vars)) optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) for epoch in range(100): model.train() total_loss = 0 for data in train_dataset: optimizer.zero_grad() cont_pred, stat_pred = model(data.x, data.edge_index, data.edge_attr) # 分离连续和状态变量 cont_true = data.x[:, :len(cont_vars)] stat_true = data.x[:, len(cont_vars):] loss = hybrid_loss(cont_pred, cont_true, stat_pred, stat_true) loss.backward() optimizer.step() total_loss += loss.item() print(f'Epoch {epoch}, Loss: {total_loss/len(train_dataset):.4f}')

4.3 异常检测推理

def detect_anomalies(model, test_sequences, threshold): anomalies = [] model.eval() with torch.no_grad(): for seq in test_sequences: data = create_graph_data(seq) cont_pred, stat_pred = model(data.x, data.edge_index, data.edge_attr) # 计算误差分数 cont_true = data.x[:, :len(cont_vars)] stat_true = data.x[:, len(cont_vars):] cont_err = F.mse_loss(cont_pred, cont_true, reduction='none').mean(1) stat_err = F.binary_cross_entropy(stat_pred, stat_true, reduction='none').mean(1) total_err = (cont_err + stat_err).item() anomalies.append(total_err > threshold) return np.array(anomalies)

5. 工程优化技巧

5.1 内存优化策略

处理大型遥测数据集时的关键技巧：

• 图结构缓存：预计算MIC矩阵和静态边关系
• 增量训练：使用DataLoader的pin_memory加速GPU传输
• 混合精度训练：启用torch.cuda.amp自动混合精度

from torch_geometric.loader import DataLoader from torch.cuda.amp import GradScaler, autocast scaler = GradScaler() train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, pin_memory=True) for epoch in range(100): for data in train_loader: optimizer.zero_grad() with autocast(): cont_pred, stat_pred = model(data.x, data.edge_index, data.edge_attr) loss = hybrid_loss(cont_pred, cont_true, stat_pred, stat_true) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()

5.2 实时检测部署

生产环境部署建议架构：

[遥测数据流] → [滑动窗口生成] → [图构建模块] → [MAG模型推理] → [异常评分] → [报警系统]

关键性能指标（在NVIDIA T4 GPU上）：

6. 模型效果评估

6.1 评估指标对比

在SMAP数据集上的性能表现：

6.2 典型异常检测结果

可视化展示模型对三种典型异常的检测效果：

1. 瞬时尖峰异常：模型能快速响应短期突变
2. 持续偏移异常：有效捕捉缓慢变化的系统偏差
3. 模式突变异常：识别变量间关系断裂的情况

import matplotlib.pyplot as plt def plot_anomalies(true_series, pred_series, anomalies): plt.figure(figsize=(12, 6)) plt.plot(true_series, label='Actual Values', color='blue') plt.plot(pred_series, label='Predicted Values', color='green', linestyle='--') anomaly_points = np.where(anomalies)[0] plt.scatter(anomaly_points, true_series[anomaly_points], color='red', label='Detected Anomalies') plt.legend() plt.title('Anomaly Detection Results') plt.xlabel('Time Step') plt.ylabel('Normalized Value') plt.show()

7. 扩展应用与迁移

7.1 自定义数据集适配

迁移到新数据集的调整要点：

1. 变量类型识别：自动区分连续/状态变量
2. MIC矩阵更新：定期重新计算变量相关性
3. 阈值自适应：动态调整基于新数据分布的阈值

def adapt_to_new_dataset(new_data_path): # 加载新数据 new_sequences, _, new_cont_vars, new_stat_vars = load_and_preprocess(new_data_path) # 增量更新MIC矩阵 new_mic_matrix = update_mic_matrix(model.mic_matrix, new_sequences) # 微调模型嵌入层 model.resize_embeddings(len(new_cont_vars) + len(new_stat_vars)) # 部分参数再训练 fine_tune_model(model, new_sequences) # 重新计算阈值 new_threshold = compute_adaptive_threshold(new_sequences) return model, new_threshold

7.2 多航天器协同监测

将模型扩展至多航天器监测场景：

class FleetMonitoringSystem: def __init__(self, spacecraft_ids): self.models = {sid: MAGModel() for sid in spacecraft_ids} self.shared_memory = {} # 存储跨航天器共享特征 def update_shared_features(self, features): # 更新航天器间共享特征 self.shared_memory.update(features) def cross_spacecraft_check(self, current_anomalies): # 基于共享特征验证异常 confirmed_anomalies = {} for sid, anomalies in current_anomalies.items(): if anomalies and self.shared_memory.get(sid+'_corroborate', False): confirmed_anomalies[sid] = anomalies return confirmed_anomalies