在工业设备维护和智能制造领域故障诊断一直是技术难点。传统方法依赖专家经验和固定阈值难以应对复杂工况下的早期故障预警。本文将基于PyTorch框架通过LSTM长短时记忆网络实现旋转机械的故障诊断实战包含数据预处理、模型构建、训练优化到实际预测的全流程代码并提供可复现的工程示例。1. LSTM故障诊断核心原理1.1 什么是LSTM网络LSTM是一种特殊的循环神经网络RNN专门设计用于处理时间序列数据中的长期依赖关系。与传统RNN相比LSTM通过引入门控机制输入门、遗忘门、输出门和细胞状态有效解决了梯度消失和爆炸问题。在故障诊断场景中设备振动信号、温度变化等时间序列数据往往包含重要的故障特征。LSTM能够捕捉这些信号中的时序模式识别出异常状态的发展趋势。1.2 LSTM在故障诊断中的优势时序特征提取自动学习振动信号中的时间相关性多变量处理同时处理多个传感器数据振动、温度、压力等早期预警通过序列模式识别实现故障早期检测自适应学习无需手动设计特征端到端学习故障特征2. 环境准备与数据说明2.1 开发环境配置# 环境要求 Python 3.8 PyTorch 1.9 NumPy 1.21 Pandas 1.3 Scikit-learn 0.24 Matplotlib 3.4 # 安装命令 pip install torch torchvision torchaudio pip install numpy pandas scikit-learn matplotlib2.2 数据集介绍本文使用凯斯西储大学CWRU轴承数据集包含正常状态和多种故障类型内圈、外圈、滚动体故障的振动信号。import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split # 数据加载示例 def load_bearing_data(data_path): 加载轴承振动数据 # 实际项目中替换为真实数据路径 data pd.read_csv(data_path) return data # 数据特征说明 采样频率12kHz 信号长度每个样本4096个点 故障类型0-正常, 1-内圈故障, 2-外圈故障, 3-滚动体故障 负载条件0HP, 1HP, 2HP, 3HP 3. 数据预处理与特征工程3.1 时序数据标准化class TimeSeriesPreprocessor: def __init__(self, sequence_length100, step_size50): self.sequence_length sequence_length self.step_size step_size self.scaler StandardScaler() def create_sequences(self, data, labels): 将时序数据转换为LSTM输入格式 sequences [] sequence_labels [] for i in range(0, len(data) - self.sequence_length, self.step_size): sequence data[i:i self.sequence_length] label labels[i self.sequence_length - 1] # 使用序列末尾的标签 sequences.append(sequence) sequence_labels.append(label) return np.array(sequences), np.array(sequence_labels) def fit_transform(self, train_data): 标准化训练数据 return self.scaler.fit_transform(train_data) def transform(self, test_data): 标准化测试数据 return self.scaler.transform(test_data) # 使用示例 preprocessor TimeSeriesPreprocessor(sequence_length100, step_size50)3.2 数据增强策略def augment_time_series(data, labels, augmentation_factor2): 时序数据增强添加噪声、时间扭曲等 augmented_data [] augmented_labels [] for sequence, label in zip(data, labels): # 原始数据 augmented_data.append(sequence) augmented_labels.append(label) # 添加高斯噪声 noise np.random.normal(0, 0.01, sequence.shape) augmented_data.append(sequence noise) augmented_labels.append(label) # 时间缩放轻微 if len(augmented_data) len(data) * augmentation_factor: scaled_sequence sequence[::2] # 下采样 if len(scaled_sequence) len(sequence) // 2: # 插值回原长度 from scipy import interpolate x_original np.arange(len(scaled_sequence)) x_new np.linspace(0, len(scaled_sequence)-1, len(sequence)) f interpolate.interp1d(x_original, scaled_sequence, kindlinear) scaled_sequence f(x_new) augmented_data.append(scaled_sequence) augmented_labels.append(label) return np.array(augmented_data), np.array(augmented_labels)4. LSTM故障诊断模型构建4.1 模型架构设计import torch import torch.nn as nn import torch.optim as optim class LSTMFaultDiagnosis(nn.Module): def __init__(self, input_size, hidden_size, num_layers, num_classes, dropout_rate0.3): super(LSTMFaultDiagnosis, self).__init__() self.hidden_size hidden_size self.num_layers num_layers # LSTM层 self.lstm nn.LSTM(input_size, hidden_size, num_layers, batch_firstTrue, dropoutdropout_rate) # 注意力机制 self.attention nn.Sequential( nn.Linear(hidden_size, hidden_size // 2), nn.Tanh(), nn.Linear(hidden_size // 2, 1), nn.Softmax(dim1) ) # 分类层 self.fc nn.Sequential( nn.Dropout(dropout_rate), nn.Linear(hidden_size, hidden_size // 2), nn.ReLU(), nn.Dropout(dropout_rate), nn.Linear(hidden_size // 2, num_classes) ) def forward(self, x): # LSTM前向传播 lstm_out, (h_n, c_n) self.lstm(x) # 注意力权重计算 attention_weights self.attention(lstm_out) context_vector torch.sum(attention_weights * lstm_out, dim1) # 分类 output self.fc(context_vector) return output # 模型参数配置 model_config { input_size: 1, # 单变量时间序列 hidden_size: 64, # LSTM隐藏层维度 num_layers: 2, # LSTM层数 num_classes: 4, # 故障类型数量 dropout_rate: 0.3 # 丢弃率 }4.2 模型初始化与验证def initialize_model(config, device): 初始化模型并移动到设备 model LSTMFaultDiagnosis( input_sizeconfig[input_size], hidden_sizeconfig[hidden_size], num_layersconfig[num_layers], num_classesconfig[num_classes], dropout_rateconfig[dropout_rate] ) return model.to(device) # 验证模型结构 device torch.device(cuda if torch.cuda.is_available() else cpu) model initialize_model(model_config, device) # 打印模型参数 def count_parameters(model): return sum(p.numel() for p in model.parameters() if p.requires_grad) print(f模型参数量: {count_parameters(model):,})5. 模型训练与优化5.1 训练流程实现class FaultDiagnosisTrainer: def __init__(self, model, train_loader, val_loader, criterion, optimizer, device): self.model model self.train_loader train_loader self.val_loader val_loader self.criterion criterion self.optimizer optimizer self.device device self.train_losses [] self.val_losses [] self.val_accuracies [] def train_epoch(self): 单个训练周期 self.model.train() running_loss 0.0 for batch_idx, (data, targets) in enumerate(self.train_loader): data, targets data.to(self.device), targets.to(self.device) # 前向传播 outputs self.model(data) loss self.criterion(outputs, targets) # 反向传播 self.optimizer.zero_grad() loss.backward() # 梯度裁剪 torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm1.0) self.optimizer.step() running_loss loss.item() return running_loss / len(self.train_loader) def validate(self): 验证集评估 self.model.eval() val_loss 0.0 correct 0 total 0 with torch.no_grad(): for data, targets in self.val_loader: data, targets data.to(self.device), targets.to(self.device) outputs self.model(data) val_loss self.criterion(outputs, targets).item() _, predicted outputs.max(1) total targets.size(0) correct predicted.eq(targets).sum().item() accuracy 100. * correct / total return val_loss / len(self.val_loader), accuracy def train(self, epochs, early_stopping_patience10): 完整训练过程 best_accuracy 0 patience_counter 0 for epoch in range(epochs): train_loss self.train_epoch() val_loss, val_accuracy self.validate() self.train_losses.append(train_loss) self.val_losses.append(val_loss) self.val_accuracies.append(val_accuracy) print(fEpoch: {epoch1:03d} | fTrain Loss: {train_loss:.4f} | fVal Loss: {val_loss:.4f} | fVal Acc: {val_accuracy:.2f}%) # 早停机制 if val_accuracy best_accuracy: best_accuracy val_accuracy patience_counter 0 # 保存最佳模型 torch.save(self.model.state_dict(), best_model.pth) else: patience_counter 1 if patience_counter early_stopping_patience: print(f早停触发最佳准确率: {best_accuracy:.2f}%) break5.2 学习率调度与优化器配置def setup_training_components(model, learning_rate0.001): 配置训练组件 # 损失函数带类别权重处理不平衡数据 criterion nn.CrossEntropyLoss() # 优化器 optimizer optim.AdamW(model.parameters(), lrlearning_rate, weight_decay1e-4) # 学习率调度器 scheduler optim.lr_scheduler.ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5, verboseTrue ) return criterion, optimizer, scheduler # 数据加载器创建 from torch.utils.data import DataLoader, TensorDataset def create_data_loaders(X_train, y_train, X_val, y_val, batch_size32): 创建PyTorch数据加载器 train_dataset TensorDataset( torch.FloatTensor(X_train).unsqueeze(-1), # 添加特征维度 torch.LongTensor(y_train) ) val_dataset TensorDataset( torch.FloatTensor(X_val).unsqueeze(-1), torch.LongTensor(y_val) ) train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader6. 模型评估与结果分析6.1 综合评估指标from sklearn.metrics import classification_report, confusion_matrix import seaborn as sns import matplotlib.pyplot as plt def evaluate_model(model, test_loader, device, class_names): 全面评估模型性能 model.eval() all_predictions [] all_targets [] with torch.no_grad(): for data, targets in test_loader: data data.to(device) outputs model(data) _, predicted outputs.max(1) all_predictions.extend(predicted.cpu().numpy()) all_targets.extend(targets.numpy()) # 分类报告 print(分类报告:) print(classification_report(all_targets, all_predictions, target_namesclass_names)) # 混淆矩阵可视化 cm confusion_matrix(all_targets, all_predictions) plt.figure(figsize(8, 6)) sns.heatmap(cm, annotTrue, fmtd, cmapBlues, xticklabelsclass_names, yticklabelsclass_names) plt.title(混淆矩阵) plt.ylabel(真实标签) plt.xlabel(预测标签) plt.show() return all_predictions, all_targets # 类别名称 class_names [正常, 内圈故障, 外圈故障, 滚动体故障]6.2 故障预警能力分析def analyze_fault_detection(predictions, targets, sequence_length): 分析故障检测的时效性 fault_indices np.where(targets ! 0)[0] # 故障样本索引 detection_delays [] for fault_idx in fault_indices: # 查找故障首次被检测到的位置 detection_idx fault_idx while detection_idx 0 and predictions[detection_idx] 0: detection_idx - 1 if predictions[detection_idx] ! 0: # 成功检测到故障 delay (fault_idx - detection_idx) * sequence_length detection_delays.append(delay) if detection_delays: avg_delay np.mean(detection_delays) print(f平均故障检测延迟: {avg_delay:.2f}个采样点) return avg_delay return 07. 实际部署与工程优化7.1 模型轻量化与加速class LightweightLSTM(nn.Module): 轻量级LSTM模型适合边缘部署 def __init__(self, input_size, hidden_size, num_classes): super(LightweightLSTM, self).__init__() self.lstm nn.LSTM(input_size, hidden_size, 1, batch_firstTrue) self.classifier nn.Linear(hidden_size, num_classes) def forward(self, x): lstm_out, _ self.lstm(x) # 取最后一个时间步的输出 last_output lstm_out[:, -1, :] return self.classifier(last_output) def convert_to_onnx(model, input_size, sequence_length, output_path): 将模型转换为ONNX格式便于部署 dummy_input torch.randn(1, sequence_length, input_size) torch.onnx.export(model, dummy_input, output_path, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}}, opset_version11)7.2 实时故障诊断流水线class RealTimeFaultDiagnosis: def __init__(self, model_path, sequence_length, device): self.sequence_length sequence_length self.device device self.model self.load_model(model_path) self.buffer [] def load_model(self, model_path): 加载训练好的模型 model LSTMFaultDiagnosis(**model_config) model.load_state_dict(torch.load(model_path, map_locationself.device)) model.eval() return model def add_data_point(self, data_point): 添加实时数据点 self.buffer.append(data_point) if len(self.buffer) self.sequence_length: self.buffer.pop(0) def predict(self): 实时预测 if len(self.buffer) self.sequence_length: return None sequence np.array(self.buffer).reshape(1, -1, 1) tensor_sequence torch.FloatTensor(sequence).to(self.device) with torch.no_grad(): output self.model(tensor_sequence) prediction output.argmax(dim1).item() return prediction def monitor_continuous(self, data_stream, window_size10): 连续监控与预警 predictions [] confidence_scores [] for data_point in data_stream: self.add_data_point(data_point) prediction self.predict() if prediction is not None: predictions.append(prediction) # 基于窗口的决策逻辑 if len(predictions) window_size: recent_predictions predictions[-window_size:] fault_count sum(1 for p in recent_predictions if p ! 0) if fault_count window_size * 0.7: # 70%的窗口内检测到故障 print(f故障预警! 故障类型: {class_names[prediction]}) return prediction return 0 # 正常状态8. 常见问题与解决方案8.1 训练过程中的典型问题问题现象可能原因解决方案损失值不下降学习率过大/过小调整学习率使用学习率调度器过拟合严重模型复杂度过高增加Dropout添加正则化数据增强梯度爆炸网络层数过深梯度裁剪使用梯度归一化类别不平衡故障样本稀少使用加权损失函数过采样技术8.2 模型性能优化技巧# 1. 学习率搜索 def find_optimal_lr(model, train_loader, device): 学习率范围测试 optimizer optim.Adam(model.parameters(), lr1e-7) lr_finder LRFinder(model, optimizer, criterion, devicedevice) lr_finder.range_test(train_loader, end_lr1, num_iter100) lr_finder.plot() # 显示最佳学习率区间 return lr_finder.best_lr # 2. 模型集成 class ModelEnsemble: 多模型集成提升鲁棒性 def __init__(self, model_paths, device): self.models [] for path in model_paths: model LSTMFaultDiagnosis(**model_config) model.load_state_dict(torch.load(path, map_locationdevice)) model.eval() self.models.append(model) def predict(self, x): predictions [] for model in self.models: with torch.no_grad(): output model(x) predictions.append(output.softmax(dim1)) # 平均概率 avg_probs torch.mean(torch.stack(predictions), dim0) return avg_probs.argmax(dim1)9. 工程实践建议9.1 生产环境部署注意事项数据质量监控实时监测传感器数据质量处理缺失值和异常值模型版本管理建立模型版本控制系统记录每次更新的性能变化故障预警阈值根据实际业务需求调整预警灵敏度回退机制当新模型性能下降时快速回退到稳定版本9.2 持续学习与模型更新class ContinuousLearning: 持续学习框架适应设备老化等变化 def __init__(self, base_model, memory_size1000): self.base_model base_model self.memory_buffer [] # 存储历史数据 self.memory_size memory_size def update_model(self, new_data, new_labels, learning_rate0.0001): 增量学习更新模型 # 保护已有知识的同时学习新模式 optimizer optim.Adam(self.base_model.parameters(), lrlearning_rate) # 结合历史数据和新数据训练 combined_data self.memory_buffer list(zip(new_data, new_labels)) if len(combined_data) self.memory_size: combined_data combined_data[-self.memory_size:] # 训练过程 self.base_model.train() for epoch in range(5): # 少量epochs避免灾难性遗忘 for data, label in combined_data: optimizer.zero_grad() output self.base_model(data.unsqueeze(0)) loss nn.CrossEntropyLoss()(output, label.unsqueeze(0)) loss.backward() optimizer.step() # 更新记忆缓冲区 self.update_memory_buffer(new_data, new_labels) def update_memory_buffer(self, new_data, new_labels): 更新记忆缓冲区优先保留稀有类别样本 for data, label in zip(new_data, new_labels): if len(self.memory_buffer) self.memory_size: self.memory_buffer.append((data, label)) else: # 替换策略随机替换或基于重要性的替换 if label in [2, 3]: # 稀有故障类型优先保留 # 替换一个常见类别的样本 common_indices [i for i, (_, l) in enumerate(self.memory_buffer) if l in [0, 1]] if common_indices: replace_idx np.random.choice(common_indices) self.memory_buffer[replace_idx] (data, label)本文完整实现了基于LSTM的故障诊断系统从数据预处理到模型部署的全流程。关键是要根据实际设备特性调整模型参数并建立持续监控机制。在实际应用中建议先在小规模数据上验证方案可行性再逐步扩展到全厂设备监控。 基于PyTorch和LSTM的工业设备故障诊断实战指南
在工业设备维护和智能制造领域故障诊断一直是技术难点。传统方法依赖专家经验和固定阈值难以应对复杂工况下的早期故障预警。本文将基于PyTorch框架通过LSTM长短时记忆网络实现旋转机械的故障诊断实战包含数据预处理、模型构建、训练优化到实际预测的全流程代码并提供可复现的工程示例。1. LSTM故障诊断核心原理1.1 什么是LSTM网络LSTM是一种特殊的循环神经网络RNN专门设计用于处理时间序列数据中的长期依赖关系。与传统RNN相比LSTM通过引入门控机制输入门、遗忘门、输出门和细胞状态有效解决了梯度消失和爆炸问题。在故障诊断场景中设备振动信号、温度变化等时间序列数据往往包含重要的故障特征。LSTM能够捕捉这些信号中的时序模式识别出异常状态的发展趋势。1.2 LSTM在故障诊断中的优势时序特征提取自动学习振动信号中的时间相关性多变量处理同时处理多个传感器数据振动、温度、压力等早期预警通过序列模式识别实现故障早期检测自适应学习无需手动设计特征端到端学习故障特征2. 环境准备与数据说明2.1 开发环境配置# 环境要求 Python 3.8 PyTorch 1.9 NumPy 1.21 Pandas 1.3 Scikit-learn 0.24 Matplotlib 3.4 # 安装命令 pip install torch torchvision torchaudio pip install numpy pandas scikit-learn matplotlib2.2 数据集介绍本文使用凯斯西储大学CWRU轴承数据集包含正常状态和多种故障类型内圈、外圈、滚动体故障的振动信号。import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler from sklearn.model_selection import train_test_split # 数据加载示例 def load_bearing_data(data_path): 加载轴承振动数据 # 实际项目中替换为真实数据路径 data pd.read_csv(data_path) return data # 数据特征说明 采样频率12kHz 信号长度每个样本4096个点 故障类型0-正常, 1-内圈故障, 2-外圈故障, 3-滚动体故障 负载条件0HP, 1HP, 2HP, 3HP 3. 数据预处理与特征工程3.1 时序数据标准化class TimeSeriesPreprocessor: def __init__(self, sequence_length100, step_size50): self.sequence_length sequence_length self.step_size step_size self.scaler StandardScaler() def create_sequences(self, data, labels): 将时序数据转换为LSTM输入格式 sequences [] sequence_labels [] for i in range(0, len(data) - self.sequence_length, self.step_size): sequence data[i:i self.sequence_length] label labels[i self.sequence_length - 1] # 使用序列末尾的标签 sequences.append(sequence) sequence_labels.append(label) return np.array(sequences), np.array(sequence_labels) def fit_transform(self, train_data): 标准化训练数据 return self.scaler.fit_transform(train_data) def transform(self, test_data): 标准化测试数据 return self.scaler.transform(test_data) # 使用示例 preprocessor TimeSeriesPreprocessor(sequence_length100, step_size50)3.2 数据增强策略def augment_time_series(data, labels, augmentation_factor2): 时序数据增强添加噪声、时间扭曲等 augmented_data [] augmented_labels [] for sequence, label in zip(data, labels): # 原始数据 augmented_data.append(sequence) augmented_labels.append(label) # 添加高斯噪声 noise np.random.normal(0, 0.01, sequence.shape) augmented_data.append(sequence noise) augmented_labels.append(label) # 时间缩放轻微 if len(augmented_data) len(data) * augmentation_factor: scaled_sequence sequence[::2] # 下采样 if len(scaled_sequence) len(sequence) // 2: # 插值回原长度 from scipy import interpolate x_original np.arange(len(scaled_sequence)) x_new np.linspace(0, len(scaled_sequence)-1, len(sequence)) f interpolate.interp1d(x_original, scaled_sequence, kindlinear) scaled_sequence f(x_new) augmented_data.append(scaled_sequence) augmented_labels.append(label) return np.array(augmented_data), np.array(augmented_labels)4. LSTM故障诊断模型构建4.1 模型架构设计import torch import torch.nn as nn import torch.optim as optim class LSTMFaultDiagnosis(nn.Module): def __init__(self, input_size, hidden_size, num_layers, num_classes, dropout_rate0.3): super(LSTMFaultDiagnosis, self).__init__() self.hidden_size hidden_size self.num_layers num_layers # LSTM层 self.lstm nn.LSTM(input_size, hidden_size, num_layers, batch_firstTrue, dropoutdropout_rate) # 注意力机制 self.attention nn.Sequential( nn.Linear(hidden_size, hidden_size // 2), nn.Tanh(), nn.Linear(hidden_size // 2, 1), nn.Softmax(dim1) ) # 分类层 self.fc nn.Sequential( nn.Dropout(dropout_rate), nn.Linear(hidden_size, hidden_size // 2), nn.ReLU(), nn.Dropout(dropout_rate), nn.Linear(hidden_size // 2, num_classes) ) def forward(self, x): # LSTM前向传播 lstm_out, (h_n, c_n) self.lstm(x) # 注意力权重计算 attention_weights self.attention(lstm_out) context_vector torch.sum(attention_weights * lstm_out, dim1) # 分类 output self.fc(context_vector) return output # 模型参数配置 model_config { input_size: 1, # 单变量时间序列 hidden_size: 64, # LSTM隐藏层维度 num_layers: 2, # LSTM层数 num_classes: 4, # 故障类型数量 dropout_rate: 0.3 # 丢弃率 }4.2 模型初始化与验证def initialize_model(config, device): 初始化模型并移动到设备 model LSTMFaultDiagnosis( input_sizeconfig[input_size], hidden_sizeconfig[hidden_size], num_layersconfig[num_layers], num_classesconfig[num_classes], dropout_rateconfig[dropout_rate] ) return model.to(device) # 验证模型结构 device torch.device(cuda if torch.cuda.is_available() else cpu) model initialize_model(model_config, device) # 打印模型参数 def count_parameters(model): return sum(p.numel() for p in model.parameters() if p.requires_grad) print(f模型参数量: {count_parameters(model):,})5. 模型训练与优化5.1 训练流程实现class FaultDiagnosisTrainer: def __init__(self, model, train_loader, val_loader, criterion, optimizer, device): self.model model self.train_loader train_loader self.val_loader val_loader self.criterion criterion self.optimizer optimizer self.device device self.train_losses [] self.val_losses [] self.val_accuracies [] def train_epoch(self): 单个训练周期 self.model.train() running_loss 0.0 for batch_idx, (data, targets) in enumerate(self.train_loader): data, targets data.to(self.device), targets.to(self.device) # 前向传播 outputs self.model(data) loss self.criterion(outputs, targets) # 反向传播 self.optimizer.zero_grad() loss.backward() # 梯度裁剪 torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm1.0) self.optimizer.step() running_loss loss.item() return running_loss / len(self.train_loader) def validate(self): 验证集评估 self.model.eval() val_loss 0.0 correct 0 total 0 with torch.no_grad(): for data, targets in self.val_loader: data, targets data.to(self.device), targets.to(self.device) outputs self.model(data) val_loss self.criterion(outputs, targets).item() _, predicted outputs.max(1) total targets.size(0) correct predicted.eq(targets).sum().item() accuracy 100. * correct / total return val_loss / len(self.val_loader), accuracy def train(self, epochs, early_stopping_patience10): 完整训练过程 best_accuracy 0 patience_counter 0 for epoch in range(epochs): train_loss self.train_epoch() val_loss, val_accuracy self.validate() self.train_losses.append(train_loss) self.val_losses.append(val_loss) self.val_accuracies.append(val_accuracy) print(fEpoch: {epoch1:03d} | fTrain Loss: {train_loss:.4f} | fVal Loss: {val_loss:.4f} | fVal Acc: {val_accuracy:.2f}%) # 早停机制 if val_accuracy best_accuracy: best_accuracy val_accuracy patience_counter 0 # 保存最佳模型 torch.save(self.model.state_dict(), best_model.pth) else: patience_counter 1 if patience_counter early_stopping_patience: print(f早停触发最佳准确率: {best_accuracy:.2f}%) break5.2 学习率调度与优化器配置def setup_training_components(model, learning_rate0.001): 配置训练组件 # 损失函数带类别权重处理不平衡数据 criterion nn.CrossEntropyLoss() # 优化器 optimizer optim.AdamW(model.parameters(), lrlearning_rate, weight_decay1e-4) # 学习率调度器 scheduler optim.lr_scheduler.ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5, verboseTrue ) return criterion, optimizer, scheduler # 数据加载器创建 from torch.utils.data import DataLoader, TensorDataset def create_data_loaders(X_train, y_train, X_val, y_val, batch_size32): 创建PyTorch数据加载器 train_dataset TensorDataset( torch.FloatTensor(X_train).unsqueeze(-1), # 添加特征维度 torch.LongTensor(y_train) ) val_dataset TensorDataset( torch.FloatTensor(X_val).unsqueeze(-1), torch.LongTensor(y_val) ) train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_dataset, batch_sizebatch_size, shuffleFalse) return train_loader, val_loader6. 模型评估与结果分析6.1 综合评估指标from sklearn.metrics import classification_report, confusion_matrix import seaborn as sns import matplotlib.pyplot as plt def evaluate_model(model, test_loader, device, class_names): 全面评估模型性能 model.eval() all_predictions [] all_targets [] with torch.no_grad(): for data, targets in test_loader: data data.to(device) outputs model(data) _, predicted outputs.max(1) all_predictions.extend(predicted.cpu().numpy()) all_targets.extend(targets.numpy()) # 分类报告 print(分类报告:) print(classification_report(all_targets, all_predictions, target_namesclass_names)) # 混淆矩阵可视化 cm confusion_matrix(all_targets, all_predictions) plt.figure(figsize(8, 6)) sns.heatmap(cm, annotTrue, fmtd, cmapBlues, xticklabelsclass_names, yticklabelsclass_names) plt.title(混淆矩阵) plt.ylabel(真实标签) plt.xlabel(预测标签) plt.show() return all_predictions, all_targets # 类别名称 class_names [正常, 内圈故障, 外圈故障, 滚动体故障]6.2 故障预警能力分析def analyze_fault_detection(predictions, targets, sequence_length): 分析故障检测的时效性 fault_indices np.where(targets ! 0)[0] # 故障样本索引 detection_delays [] for fault_idx in fault_indices: # 查找故障首次被检测到的位置 detection_idx fault_idx while detection_idx 0 and predictions[detection_idx] 0: detection_idx - 1 if predictions[detection_idx] ! 0: # 成功检测到故障 delay (fault_idx - detection_idx) * sequence_length detection_delays.append(delay) if detection_delays: avg_delay np.mean(detection_delays) print(f平均故障检测延迟: {avg_delay:.2f}个采样点) return avg_delay return 07. 实际部署与工程优化7.1 模型轻量化与加速class LightweightLSTM(nn.Module): 轻量级LSTM模型适合边缘部署 def __init__(self, input_size, hidden_size, num_classes): super(LightweightLSTM, self).__init__() self.lstm nn.LSTM(input_size, hidden_size, 1, batch_firstTrue) self.classifier nn.Linear(hidden_size, num_classes) def forward(self, x): lstm_out, _ self.lstm(x) # 取最后一个时间步的输出 last_output lstm_out[:, -1, :] return self.classifier(last_output) def convert_to_onnx(model, input_size, sequence_length, output_path): 将模型转换为ONNX格式便于部署 dummy_input torch.randn(1, sequence_length, input_size) torch.onnx.export(model, dummy_input, output_path, input_names[input], output_names[output], dynamic_axes{input: {0: batch_size}}, opset_version11)7.2 实时故障诊断流水线class RealTimeFaultDiagnosis: def __init__(self, model_path, sequence_length, device): self.sequence_length sequence_length self.device device self.model self.load_model(model_path) self.buffer [] def load_model(self, model_path): 加载训练好的模型 model LSTMFaultDiagnosis(**model_config) model.load_state_dict(torch.load(model_path, map_locationself.device)) model.eval() return model def add_data_point(self, data_point): 添加实时数据点 self.buffer.append(data_point) if len(self.buffer) self.sequence_length: self.buffer.pop(0) def predict(self): 实时预测 if len(self.buffer) self.sequence_length: return None sequence np.array(self.buffer).reshape(1, -1, 1) tensor_sequence torch.FloatTensor(sequence).to(self.device) with torch.no_grad(): output self.model(tensor_sequence) prediction output.argmax(dim1).item() return prediction def monitor_continuous(self, data_stream, window_size10): 连续监控与预警 predictions [] confidence_scores [] for data_point in data_stream: self.add_data_point(data_point) prediction self.predict() if prediction is not None: predictions.append(prediction) # 基于窗口的决策逻辑 if len(predictions) window_size: recent_predictions predictions[-window_size:] fault_count sum(1 for p in recent_predictions if p ! 0) if fault_count window_size * 0.7: # 70%的窗口内检测到故障 print(f故障预警! 故障类型: {class_names[prediction]}) return prediction return 0 # 正常状态8. 常见问题与解决方案8.1 训练过程中的典型问题问题现象可能原因解决方案损失值不下降学习率过大/过小调整学习率使用学习率调度器过拟合严重模型复杂度过高增加Dropout添加正则化数据增强梯度爆炸网络层数过深梯度裁剪使用梯度归一化类别不平衡故障样本稀少使用加权损失函数过采样技术8.2 模型性能优化技巧# 1. 学习率搜索 def find_optimal_lr(model, train_loader, device): 学习率范围测试 optimizer optim.Adam(model.parameters(), lr1e-7) lr_finder LRFinder(model, optimizer, criterion, devicedevice) lr_finder.range_test(train_loader, end_lr1, num_iter100) lr_finder.plot() # 显示最佳学习率区间 return lr_finder.best_lr # 2. 模型集成 class ModelEnsemble: 多模型集成提升鲁棒性 def __init__(self, model_paths, device): self.models [] for path in model_paths: model LSTMFaultDiagnosis(**model_config) model.load_state_dict(torch.load(path, map_locationdevice)) model.eval() self.models.append(model) def predict(self, x): predictions [] for model in self.models: with torch.no_grad(): output model(x) predictions.append(output.softmax(dim1)) # 平均概率 avg_probs torch.mean(torch.stack(predictions), dim0) return avg_probs.argmax(dim1)9. 工程实践建议9.1 生产环境部署注意事项数据质量监控实时监测传感器数据质量处理缺失值和异常值模型版本管理建立模型版本控制系统记录每次更新的性能变化故障预警阈值根据实际业务需求调整预警灵敏度回退机制当新模型性能下降时快速回退到稳定版本9.2 持续学习与模型更新class ContinuousLearning: 持续学习框架适应设备老化等变化 def __init__(self, base_model, memory_size1000): self.base_model base_model self.memory_buffer [] # 存储历史数据 self.memory_size memory_size def update_model(self, new_data, new_labels, learning_rate0.0001): 增量学习更新模型 # 保护已有知识的同时学习新模式 optimizer optim.Adam(self.base_model.parameters(), lrlearning_rate) # 结合历史数据和新数据训练 combined_data self.memory_buffer list(zip(new_data, new_labels)) if len(combined_data) self.memory_size: combined_data combined_data[-self.memory_size:] # 训练过程 self.base_model.train() for epoch in range(5): # 少量epochs避免灾难性遗忘 for data, label in combined_data: optimizer.zero_grad() output self.base_model(data.unsqueeze(0)) loss nn.CrossEntropyLoss()(output, label.unsqueeze(0)) loss.backward() optimizer.step() # 更新记忆缓冲区 self.update_memory_buffer(new_data, new_labels) def update_memory_buffer(self, new_data, new_labels): 更新记忆缓冲区优先保留稀有类别样本 for data, label in zip(new_data, new_labels): if len(self.memory_buffer) self.memory_size: self.memory_buffer.append((data, label)) else: # 替换策略随机替换或基于重要性的替换 if label in [2, 3]: # 稀有故障类型优先保留 # 替换一个常见类别的样本 common_indices [i for i, (_, l) in enumerate(self.memory_buffer) if l in [0, 1]] if common_indices: replace_idx np.random.choice(common_indices) self.memory_buffer[replace_idx] (data, label)本文完整实现了基于LSTM的故障诊断系统从数据预处理到模型部署的全流程。关键是要根据实际设备特性调整模型参数并建立持续监控机制。在实际应用中建议先在小规模数据上验证方案可行性再逐步扩展到全厂设备监控。