算法首先通过滑动窗口技术将原始振动时序数据分割成固定长度的片段，然后利用基于神经随机微分方程NSDE的编码器对数据进行深度特征提取，捕捉时间序列的随机动态特性。接着通过长短期多分支自编码器LST-AE对特征进行重构学习，分别提取数据的长期依赖关系和短期波动模式，并基于重构误差自动生成伪标签区分正常与异常状态。最后结合对比学习技术，通过正负样本对比优化特征空间，训练分类器准确识别滚珠故障、内圈故障、外圈故障等多种故障类型，形成了一套从无监督特征学习到半监督故障诊断的完整解决方案。
算法详细步骤：
首先对原始振动信号进行标准化预处理，消除量纲影响，然后采用滑动窗口技术将连续时间序列切割成固定长度的数据片段。使用神经随机微分方程编码器对每个数据窗口进行深度特征提取，该编码器能够同时建模确定性漂移和随机扩散过程，有效捕捉振动信号的动态特性。
将提取的特征输入长短期多分支自编码器，分别构建长期特征学习分支、短期特征学习分支和原始特征学习分支。长期分支关注整体趋势和周期性模式，短期分支聚焦近期波动和瞬态特征，三个分支并行进行特征重构，通过比较重构误差挖掘数据的多尺度信息。
基于多个分支的重构误差计算综合异常得分，设定自适应阈值区分正常与故障状态，自动生成伪标签数据。这个过程完全无监督，仅需少量正常样本即可建立准确的异常检测基准。
最后结合对比学习策略，将提取的特征投影到低维空间，通过最大化同类样本相似度、最小化异类样本相似度的方式优化特征表示。同时利用生成的伪标签和少量真实标签训练多分类器，最终实现对滚动轴承正常状态、滚珠故障、内圈故障和外圈故障四种状态的精准识别。整个过程形成从数据预处理、特征学习、伪标签生成到分类决策的完整诊断流程。
import torch import torch.nn as nn import torch.nn.functional as F import torchdiffeq import numpy as np import matplotlib.pyplot as plt from sklearn.preprocessing import StandardScaler from sklearn.metrics import confusion_matrix, classification_report, accuracy_score import seaborn as sns import pandas as pd from pathlib import Path import warnings warnings.filterwarnings('ignore') # 设备配置 device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') print(f"Using device: {device}") class EMA_NSDE_Encoder(nn.Module): """ 基于神经随机微分方程的编码器 修复了维度问题 """ def __init__(self, input_dim=1, hidden_dim=128, output_dim=128): super().__init__() self.input_dim = input_dim self.hidden_dim = hidden_dim self.output_dim = output_dim # 输入投影层：将输入维度映射到隐藏维度 self.input_projection = nn.Linear(input_dim, hidden_dim) # 漂移项网络 - 输入是隐藏状态+时间，输出是隐藏状态的变化率 self.drift_net = nn.Sequential( nn.Linear(hidden_dim + 1, 256), # +1 for time nn.Tanh(), nn.Linear(256, 256), nn.Tanh(), nn.Linear(256, hidden_dim) ) # 扩散项网络 self.diffusion_net = nn.Sequential( nn.Linear(hidden_dim + 1, 256), # +1 for time nn.Tanh(), nn.Linear(256, 256), nn.Tanh(), nn.Linear(256, hidden_dim) ) def forward(self, t, z): """计算SDE的漂移和扩散项""" # z的形状应该是 [batch_size, hidden_dim] batch_size = z.shape[0] # 添加时间维度 t_expanded = torch.ones(batch_size, 1, device=z.device) * t # 拼接时间和隐状态 z_with_time = torch.cat([z, t_expanded], dim=1) # 计算漂移项和扩散项 drift = self.drift_net(z_with_time) diffusion = self.diffusion_net(z_with_time) return drift, diffusion * 0.05 # 减小扩散项的强度 def encode(self, x, t_span): """编码输入序列""" # x: [batch_size, seq_len, input_dim] batch_size, seq_len, input_dim = x.shape # 初始状态 - 对第一个时间步进行投影 first_step = x[:, 0, :] # [batch_size, input_dim] z0 = self.input_projection(first_step) # [batch_size, hidden_dim] # 使用欧拉方法求解SDE z = z0 dt = (t_span[-1] - t_span[0]) / len(t_span) for t in t_span: drift, diffusion = self.forward(t, z) # 欧拉-丸山方法 noise = torch.randn_like(diffusion) z = z + drift * dt + diffusion * torch.sqrt(torch.tensor(dt, device=x.device)) * noise return z参考文章：
基于神经微分方程和对比学习的长短期多分支自编码器机械故障诊断算法（Pytorch） - 哥廷根数学学派的文章
https://zhuanlan.zhihu.com/p/1985748959298360418
工学博士，担任《Mechanical System and Signal Processing》审稿专家，担任
《中国电机工程学报》优秀审稿专家，《控制与决策》，《系统工程与电子技术》，《电力系统保护与控制》，《宇航学报》等EI期刊审稿专家。
擅长领域：现代信号处理，机器学习，深度学习，数字孪生，时间序列分析，设备缺陷检测、设备异常检测、设备智能故障诊断与健康管理PHM等。
