航天器遥测数据异常检测实战基于 PyTorch Geometric 复现 MAG 模型核心模块航天器在轨运行期间产生的遥测数据如同人体的生命体征包含着反映系统健康状态的丰富信息。这些数据通常呈现多维、长周期、非线性耦合等复杂特性传统基于统计或单一时间序列分析的方法往往难以捕捉其深层模式。本文将带您用 PyTorch GeometricPyG实现 MAGMaximum Information Coefficient Attention Graph Network模型的核心组件该模型通过图神经网络融合变量间的长期相关性和短期交互特征在 NASA 公开数据集上实现了 98.7% 的异常检测准确率。1. 环境准备与数据预处理1.1 安装依赖库确保已安装 PyTorch 1.8 和 CUDA 11.x然后安装必要库pip install torch-geometric minepy scikit-learn pip install torch-scatter torch-sparse -f https://data.pyg.org/whl/torch-1.10.0cu113.html1.2 数据加载与滑动窗口处理航天器遥测数据通常为多维时间序列我们首先进行标准化和窗口分割import numpy as np from sklearn.preprocessing import MinMaxScaler def create_sliding_windows(data, window_size50, stride1): windows [] for i in range(0, len(data) - window_size 1, stride): windows.append(data[i:iwindow_size]) return np.array(windows) # 示例加载NASA SMAP数据集 (形状为 [timesteps, features]) raw_data np.load(smap.npy) scaler MinMaxScaler() normalized_data scaler.fit_transform(raw_data) # 创建滑动窗口 (输出形状为 [num_windows, window_size, num_features]) windows create_sliding_windows(normalized_data, window_size50)2. 构建变量相关性图结构2.1 计算最大信息系数MICMIC 能有效捕捉非线性相关性我们使用 minepy 库计算from minepy import MINE def compute_mic_matrix(data, alpha0.6): n_features data.shape[1] mic_matrix np.zeros((n_features, n_features)) mine MINE(alphaalpha, c15) for i in range(n_features): for j in range(i1, n_features): mine.compute_score(data[:,i], data[:,j]) mic_matrix[i,j] mic_matrix[j,i] mine.mic() return mic_matrix # 在整个训练集上计算MIC矩阵 mic_matrix compute_mic_matrix(normalized_data)2.2 构建图数据对象将 MIC 矩阵转化为 PyG 的 Data 对象包含节点和边特征import torch from torch_geometric.data import Data def build_graph_data(window, mic_matrix, threshold0.3): num_nodes window.shape[1] # 特征维度作为节点数 # 边构造 (基于MIC阈值) edge_index [] edge_attr [] for i in range(num_nodes): for j in range(i1, num_nodes): if mic_matrix[i,j] threshold: edge_index.append([i, j]) edge_attr.append(mic_matrix[i,j]) edge_index torch.tensor(edge_index).t().contiguous() edge_attr torch.tensor(edge_attr).float().unsqueeze(1) # 节点特征 (LSTM时序特征 嵌入向量) node_features extract_temporal_features(window) # 后续实现 return Data(xnode_features, edge_indexedge_index, edge_attredge_attr)3. 时序特征提取与图卷积实现3.1 双向LSTM时序编码使用 LSTM 捕捉每个变量的时间模式class TemporalEncoder(nn.Module): def __init__(self, input_dim, hidden_dim64): super().__init__() self.lstm nn.LSTM( input_sizeinput_dim, hidden_sizehidden_dim, bidirectionalTrue, batch_firstTrue ) def forward(self, x): # x形状: [batch_size, seq_len, num_features] outputs, _ self.lstm(x) return outputs[:,-1,:] # 取最后时间步的输出3.2 注意力边权重增强在原始 MIC 边权重上加入注意力机制class EdgeAttention(nn.Module): def __init__(self, node_dim): super().__init__() self.attn nn.Sequential( nn.Linear(node_dim * 2, 1), nn.LeakyReLU() ) def forward(self, x, edge_index): row, col edge_index x_i, x_j x[row], x[col] alpha self.attn(torch.cat([x_i, x_j], dim-1)) return torch.sigmoid(alpha) # 注意力系数[0,1]3.3 图卷积层实现消息传递与信息聚合的关键组件from torch_geometric.nn import MessagePassing class MAGConv(MessagePassing): def __init__(self, node_dim, edge_dim): super().__init__(aggrmean) self.edge_encoder nn.Linear(edge_dim, node_dim) self.message_net nn.Sequential( nn.Linear(node_dim * 2, node_dim), nn.ReLU() ) def forward(self, x, edge_index, edge_attr): edge_emb self.edge_encoder(edge_attr) return self.propagate(edge_index, xx, edge_embedge_emb) def message(self, x_i, x_j, edge_emb): return self.message_net(torch.cat([x_i, x_j * edge_emb], dim-1))4. 完整模型架构与训练流程4.1 MAG 模型集成组合所有组件构建端到端模型class MAGModel(nn.Module): def __init__(self, num_features, hidden_dim128): super().__init__() self.embedding nn.Embedding(num_features, hidden_dim) self.temporal_enc TemporalEncoder(1, hidden_dim) self.edge_attn EdgeAttention(hidden_dim) self.conv1 MAGConv(hidden_dim, 1) self.conv2 MAGConv(hidden_dim, 1) self.predictor nn.Linear(hidden_dim, 1) def forward(self, data): # 节点特征: 嵌入向量 LSTM特征 node_ids torch.arange(data.x.size(0)).to(data.x.device) h_embed self.embedding(node_ids) h_temp self.temporal_enc(data.x.unsqueeze(-1)) h h_embed h_temp # 图卷积 edge_weight self.edge_attn(h, data.edge_index) h self.conv1(h, data.edge_index, data.edge_attr * edge_weight) h torch.relu(h) h self.conv2(h, data.edge_index, data.edge_attr * edge_weight) return self.predictor(h)4.2 无监督训练策略采用预测误差作为异常分数def train_epoch(model, dataloader, optimizer): model.train() total_loss 0 for batch in dataloader: optimizer.zero_grad() pred model(batch) target batch.x.mean(dim1) # 使用窗口均值作为预测目标 loss F.mse_loss(pred.squeeze(), target) loss.backward() optimizer.step() total_loss loss.item() return total_loss / len(dataloader) # 异常分数计算 def compute_anomaly_score(model, dataloader): model.eval() scores [] with torch.no_grad(): for batch in dataloader: pred model(batch) target batch.x.mean(dim1) score F.mse_loss(pred.squeeze(), target, reductionnone) scores.append(score.cpu().numpy()) return np.concatenate(scores)5. 工程优化技巧5.1 计算效率优化MIC 矩阵缓存预计算并存储 MIC 矩阵避免每次训练重复计算稀疏图处理对边进行阈值过滤保留 top-k 强连接def sparsify_mic_matrix(mic_matrix, topk10): adj np.zeros_like(mic_matrix) for i in range(len(mic_matrix)): idx np.argpartition(mic_matrix[i], -topk)[-topk:] adj[i, idx] mic_matrix[i, idx] return adj5.2 超参数选择建议参数推荐值作用窗口大小50-100平衡时序特征捕捉与计算开销MIC阈值0.3-0.5控制图连接稀疏度LSTM隐藏层64-128时序编码容量图卷积层数2-3信息传播深度学习率1e-3Adam优化器初始值5.3 多模态数据融合对于同时包含连续值和离散命令的遥测数据可扩展模型class MultiModalMAG(nn.Module): def __init__(self, cont_dim, disc_dim): super().__init__() self.cont_encoder TemporalEncoder(cont_dim) self.disc_encoder nn.Embedding(disc_dim, 32) # 其余组件类似... def forward(self, data): h_cont self.cont_encoder(data.cont_x) h_disc self.disc_encoder(data.disc_x).mean(dim1) h torch.cat([h_cont, h_disc], dim-1) # 后续图卷积...提示实际部署时建议使用 TorchScript 将模型转换为脚本模式可获得 20-30% 的性能提升