企业官网建设流程全解析

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

航天器在轨运行期间产生的遥测数据如同人体的生命体征，包含着反映系统健康状态的丰富信息。这些数据通常呈现多维、长周期、非线性耦合等复杂特性，传统基于统计或单一时间序列分析的方法往往难以捕捉其深层模式。本文将带您用 PyTorch Geometric（PyG）实现 MAG（Maximum 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.0+cu113.html

1.2 数据加载与滑动窗口处理

航天器遥测数据通常为多维时间序列，我们首先进行标准化和窗口分割：

import numpy as np from sklearn.preprocessing import MinMaxScaler def create_sliding_windows(data, window_size=50, stride=1): windows = [] for i in range(0, len(data) - window_size + 1, stride): windows.append(data[i:i+window_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_size=50)

2. 构建变量相关性图结构

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

MIC 能有效捕捉非线性相关性，我们使用 minepy 库计算：

from minepy import MINE def compute_mic_matrix(data, alpha=0.6): n_features = data.shape[1] mic_matrix = np.zeros((n_features, n_features)) mine = MINE(alpha=alpha, c=15) for i in range(n_features): for j in range(i+1, 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, threshold=0.3): num_nodes = window.shape[1] # 特征维度作为节点数 # 边构造 (基于MIC阈值) edge_index = [] edge_attr = [] for i in range(num_nodes): for j in range(i+1, 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(x=node_features, edge_index=edge_index, edge_attr=edge_attr)

3. 时序特征提取与图卷积实现

3.1 双向LSTM时序编码

使用 LSTM 捕捉每个变量的时间模式：

class TemporalEncoder(nn.Module): def __init__(self, input_dim, hidden_dim=64): super().__init__() self.lstm = nn.LSTM( input_size=input_dim, hidden_size=hidden_dim, bidirectional=True, batch_first=True ) 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__(aggr='mean') 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, x=x, edge_emb=edge_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_dim=128): 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(dim=1) # 使用窗口均值作为预测目标 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(dim=1) score = F.mse_loss(pred.squeeze(), target, reduction='none') scores.append(score.cpu().numpy()) return np.concatenate(scores)

5. 工程优化技巧

5.1 计算效率优化

• MIC 矩阵缓存：预计算并存储 MIC 矩阵，避免每次训练重复计算
• 稀疏图处理：对边进行阈值过滤，保留 top-k 强连接

def sparsify_mic_matrix(mic_matrix, topk=10): 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 adj

5.2 超参数选择建议

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(dim=1) h = torch.cat([h_cont, h_disc], dim=-1) # 后续图卷积...

提示：实际部署时建议使用 TorchScript 将模型转换为脚本模式，可获得 20-30% 的性能提升