Python深度学习机械设备故障诊断：振动信号处理与1D-CNN实战

简介：面向机械故障诊断与Python深度学习开发者的完整源码包，以多层感知机与卷积神经网络混合模型处理滚动轴承、齿轮箱振动频谱数据，实现故障类型智能识别，适用于毕业设计、课程实践及工业智能运维研发。资源共40个文件，含12个py脚本、12个xml配置、4个pyc编译文件、6个zbak工程备份，以及docx技术文档和txt说明，压缩包整体235KB，轻量易用，便于完整下载与本地复现。工程按RNN、CNN、SAE、MLP等模块划分，覆盖数据预处理、特征提取、模型训练与评估全流程，并附深度学习在故障诊断中的应用研究文档和备份代码，方便对照阅读与二次开发。已有66人学习下载，代码通过单元测试与集成验证，关键模块设有异常处理，能够为机械设备预测性维护和故障根因分析提供直接参考。

1. 机械设备故障诊断系统：为什么“听声音”值得用 Python 深度学习重做一遍

机械设备故障诊断系统这几年在工厂里越来越常见，核心做法就是把电机、轴承、齿轮箱的振动信号采集下来，用 Python 做预处理，再用深度学习模型自动判断设备是正常还是已经出现磨损、断齿、不对中。很多人以为难点在模型结构，其实我见过更多团队在数据切分和标签对齐上翻车：训练集准确率 98%，换一台设备直接掉到 70%。这套流程适合要做预测性维护的运维工程师、刚转工业算法方向的开发，以及想从源码层把整个数据流吃透的同学。下面按“数据准备 → 模型 → 源码解析 → 避坑 → 现场验证”的顺序，把一套能落地的最小方案讲清楚。

2. 把振动信号变成模型能吃的样本：滑动窗口、FFT 与切分雷区

机械设备故障诊断系统的输入通常是一维振动波形。加速度传感器以固定采样率采集信号，比如 25600 Hz，意味着每秒记录 25600 个点。原始波形很长，不可能整段塞进网络，第一步是把连续信号切成固定长度的“样本”，这个过程叫滑动窗口。切得好不好，直接决定后面模型能不能收敛。网上 python 教程里很少讲这一层，因为图像任务不需要自己切窗口，而振动数据恰恰卡在最前面。

2.1 滑动窗口切分：window_size、step_size 与重叠比例怎么定

先看一段最基础的切分代码，把一维信号变成 (N, window_size) 的样本数组。

import numpy as np def sliding_window(signal, window_size, step_size): """把一维振动信号切成多个固定长度样本。 Args: signal: 一维 numpy 数组，原始振动波形 window_size: 每个样本包含的采样点数，例如 10240 step_size: 每次滑动的采样点数，例如 5120 Returns: samples: 形状为 (N, window_size) 的二维数组 """ samples = [] for start in range(0, len(signal) - window_size + 1, step_size): samples.append(signal[start:start + window_size]) return np.array(samples)

这段代码把一段连续信号变成 N 个等长片段。range 的终点用len(signal) - window_size + 1，保证最后一个窗口不越界。step_size 小于 window_size 时，相邻窗口有重叠，样本量变大；重叠太多会让样本高度相关，我一般先设 50% 重叠，也就是 step_size 取 window_size 的一半，作为起点。

window_size 怎么定，要看设备转速和采样率。常见做法是让一个窗口至少包含 5～10 个旋转周期。例如电机转速 1500 rpm，转一圈是 40 ms；采样率 25600 Hz 时，一个周期约 1024 点，10 个周期就是 10240 点。如果窗口只取 1024 点，相当于只看一圈，故障特征可能还没完整出现，模型容易把单圈噪声当成故障模式。

2.2 频谱特征提取：采样率、FFT 分辨率和汉宁窗的作用

滑动窗口切出的是时域波形。很多故障在时域上不明显，比如轴承内圈早期点蚀，时域上只是几个小冲击，转到频域却能清楚看到故障特征频率及其边带。所以传统诊断流程里，FFT 几乎是必经步骤。深度学习模型可以直接吃原始波形，但保留一个频谱提取函数，对后面做特征对比、现场排查都很有用。

import numpy as np def extract_fft_spectrum(signal, sampling_rate): """对单个窗口做 FFT，返回幅值谱与频率轴。 Args: signal: 一维时域信号，长度 N sampling_rate: 采样率，单位 Hz Returns: freqs: 频率轴，单位 Hz magnitudes: 幅值谱 """ n = len(signal) window = np.hanning(n) spectrum = np.fft.rfft(signal * window) magnitudes = np.abs(spectrum) freqs = np.fft.rfftfreq(n, d=1.0 / sampling_rate) return freqs, magnitudes

先乘汉宁窗再做 FFT，是为了减少频谱泄漏。直接截断一段信号，等价于给信号乘了一个矩形窗，频谱旁瓣会拖得很长；汉宁窗把两端压到接近零，主瓣变宽一点，但旁瓣压下去很多，故障特征频率更容易辨认。rfft只计算实信号的正频率部分，输出点数约为 n/2 + 1，rfftfreq返回对应的频率刻度，两者一一对应。

频率分辨率等于采样率除以窗口长度。采样率 25600 Hz、窗口 10240 点时，分辨率约 2.5 Hz，要看几十赫兹的轴承特征频率足够了；如果窗口只有 1024 点，分辨率变成 25 Hz，两条靠得近的边带直接糊在一起。所以现场做频谱分析时，宁可用长窗口降低分辨率，也别贪快。

2.3 切分训练/验证集：按设备切而不是按样本切

样本切好之后，下一个关键动作是划分训练集和验证集。很多人直接从切好的样本里随机抽 70% 训练、30% 验证，这是整套系统里最隐蔽的雷区。同一台设备连续采集的信号，相邻窗口高度相似，随机切分等于把几乎重复的数据同时放进训练和验证，验证指标会虚高到不真实。

import numpy as np def split_by_machine(samples, labels, machine_ids, train_ratio=0.7): """按设备编号划分训练/验证集，避免同一台设备的数据同时出现在两边。 Args: samples: (N, L) 样本数组 labels: (N, ) 标签 machine_ids: (N, ) 每个样本所属设备编号 train_ratio: 划分到的设备中，训练集设备占比 Returns: train_samples, train_labels, val_samples, val_labels """ unique_ids = np.unique(machine_ids) n_train_machines = int(len(unique_ids) * train_ratio) train_ids = set(unique_ids[:n_train_machines]) train_mask = np.array([m in train_ids for m in machine_ids]) return (samples[train_mask], labels[train_mask], samples[~train_mask], labels[~train_mask])

这段代码先取设备编号去重，再按设备粒度划分。注意unique_ids[:n_train_machines]取前几个编号前，应该先np.random.shuffle(unique_ids)，否则设备编号有序会让某些类别全落进训练集或验证集。验证集里只要出现没见过的设备，里面样本再像，模型也没抄过答案，准确率才有参考价值。

我之前踩过一次：同一段轴承数据随机切分，验证准确率 99%，换到另一台同型号电机上直接掉到 76%。原因就是训练集和验证集里混着同一台设备的相邻窗口，模型背下了那台设备的噪声底。按设备切分之后，验证集准确率降到 92%，但现场测试稳定很多。做这套系统，数据切分这步省不得。

3. 从波形到诊断结果：用 PyTorch 搭一个能跑的 1D-CNN

从深度学习入门阶段过来的人，习惯把图像那套思路直接搬过来。但振动信号是典型的一维时序数据，模型结构和输入组织都得改。很多人找深度学习实战项目案例时，第一反应是做图像分类，其实振动数据分类更贴近工业现场，而且样本量通常比图像小得多，模型不需要很深就能出效果。

3.1 为什么 1D-CNN 是振动信号诊断的可靠基线

振动信号里的故障模式，比如轴承外圈剥落产生的冲击，在时域上是局部波形，在频域上是集中在某几个频段的能量。一维卷积核沿着时间轴滑动，天然适合捕捉这种局部模式；卷积核参数共享，又让它比全连接网络抗过拟合。相比 LSTM、Transformer，1D-CNN 参数少、训练快、部署容易，对机械设备故障诊断这种样本量不一定很大的场景，是性价比最高的起点。

我一般把 1D-CNN 当基线模型，先跑通再决定要不要上更复杂的结构。如果故障特征本身在频谱上分得很开，CNN 几层就能学到；如果样本量很大、故障形态复杂，再考虑加注意力或者换成 2D 方式把时频图喂给 ResNet。基线模型的意义是给后续所有改动定一个参照点。

3.2 搭一个 1D-CNN：网络结构、参数与输入输出形状

下面这个网络是我常用的最小结构，输入是 (batch, 1, window_size)，输出是类别得分。

import torch.nn as nn class FaultCNN1D(nn.Module): """输入形状 (batch, 1, window_size)，输出未归一化的类别得分。""" def __init__(self, num_classes=4, base_channels=16): super().__init__() self.features = nn.Sequential( nn.Conv1d(1, base_channels, kernel_size=7, stride=2, padding=3), nn.BatchNorm1d(base_channels), nn.ReLU(inplace=True), nn.MaxPool1d(kernel_size=2, stride=2), nn.Conv1d(base_channels, base_channels * 2, kernel_size=5, stride=2, padding=2), nn.BatchNorm1d(base_channels * 2), nn.ReLU(inplace=True), nn.MaxPool1d(kernel_size=2, stride=2), nn.Conv1d(base_channels * 2, base_channels * 4, kernel_size=3, stride=1, padding=1), nn.BatchNorm1d(base_channels * 4), nn.ReLU(inplace=True), ) self.classifier = nn.Sequential( nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Linear(base_channels * 4, num_classes), ) def forward(self, x): return self.classifier(self.features(x))

卷积核从 7 逐渐缩到 3，前层看更宽的波形模式，后层聚焦细节。stride=2 在卷积层里直接降采样，后面跟 MaxPool 再降一次，感受野增长快，计算量小。padding 取 kernel_size 整除 2，配合 stride=2 时，每层输出长度近似减半，不会越卷越短到负数。BatchNorm 放在卷积和激活之间，对小批量训练很关键，能压住梯度波动。

最后一层用AdaptiveAvgPool1d(1)把特征压成 1 个点再接线性层。这样有个额外好处：如果现场换了一种窗口长度，只要特征图长度不为零，模型结构不用改。Linear 输入维度固定为base_channels * 4，与 window_size 无关。

3.3 训练与评估：loss 选择、混淆矩阵和查全率

模型定义好之后，训练循环的写法直接决定能不能复现。下面是最小可跑的 epoch 训练函数。

import torch def train_one_epoch(model, dataloader, optimizer, criterion, device): """跑完一个 epoch，返回平均训练损失。""" model.train() total_loss = 0.0 num_samples = 0 for batch_x, batch_y in dataloader: batch_x = batch_x.unsqueeze(1).to(device) # (B,1,L) batch_y = batch_y.to(device) optimizer.zero_grad() outputs = model(batch_x) loss = criterion(outputs, batch_y) loss.backward() optimizer.step() total_loss += loss.item() * batch_x.size(0) num_samples += batch_x.size(0) return total_loss / num_samples

unsqueeze(1)给每个 batch 加一维 channel，因为Conv1d期望输入是 (B, C, L)。损失函数用torch.nn.CrossEntropyLoss()，它内部已经做了 softmax，模型输出层不需要再接。优化器用 Adam，学习率从 1e-3 起步；如果训练损失震荡明显，降到 3e-4 再跑。

评估不能只看准确率，尤其故障诊断这种正负样本可能严重不均衡的任务。用 sklearn 直接出分类报告和混淆矩阵，比对着 loss 曲线猜要直观得多。

from sklearn.metrics import classification_report, confusion_matrix def evaluate(model, dataloader, device): """返回预测标签与真实标签，用于后续计算指标。""" model.eval() preds, trues = [], [] with torch.no_grad(): for batch_x, batch_y in dataloader: batch_x = batch_x.unsqueeze(1).to(device) outputs = model(batch_x) preds.extend(outputs.argmax(dim=1).cpu().numpy()) trues.extend(batch_y.numpy()) return trues, preds

拿到 trues 和 preds 之后，先看classification_report里每个类别的查准率、查全率、F1，再看confusion_matrix里哪些类别互相混淆。振动诊断里我一般最关心两类指标：正常样本被误报成故障的比例，以及故障样本被漏报的比例。前者决定现场会不会三天两头响假警报，后者决定真正出问题时能不能兜住。

4. 源码解析：把故障诊断系统的模块边界与训练主流程拆开看

源码解析这部分，不打算逐行贴长文件，而是按模块边界讲清楚每个文件负责什么、数据怎么流动，再拆训练主流程里最容易写错的 checkpoint 保存和早停逻辑。一套能维护的故障诊断系统，代码结构比模型结构更重要。临时脚本能跑通 demo，但参数一改就崩，现场根本留不住。

4.1 项目模块划分：数据加载、特征、模型、训练、评估与推理

我一般会把项目拆成下面这些模块，每个文件只干一件事。

数据流是单向的：原始信号 → features.py 切窗 → data_loader.py 读成样本 → train.py 训练 → evaluate.py 评估 → infer.py 部署。特征模块和训练模块不互相 import，data_loader 只负责把切好的样本喂给模型。这样后续换模型、换特征、换数据格式，都不需要牵一发动全身。

4.2 训练主流程源码：checkpoint 保存与早停逻辑

训练主流程里最容易写错的，不是模型前向，而是模型保存时机。常见错误是每个 epoch 都保存一次，最后磁盘被 checkpoints 塞满；或者只在最后一个 epoch 保存，结果过拟合的模型覆盖了最好的参数。我习惯只保存验证集上表现最好的那个 checkpoint，配合早停逻辑。

class EarlyStopping: """验证指标连续不提升就提前停，避免无效训练。""" def __init__(self, patience=15, min_delta=0.001): self.patience = patience self.min_delta = min_delta self.counter = 0 self.best_score = None def should_stop(self, val_acc): if self.best_score is None: self.best_score = val_acc return False if val_acc > self.best_score + self.min_delta: self.best_score = val_acc self.counter = 0 else: self.counter += 1 return self.counter >= self.patience

min_delta是容忍噪声的阈值。验证准确率提升不足 0.001 也当成没进步，否则一点点波动就重置 counter，早停形同虚设。patience设 15 表示连续 15 个 epoch 没有明显提升就停。对振动数据来说，100 个 epoch 里前 20 个通常已经决定 80% 的效果，后 50 个 epoch 基本在磨验证集上最后几个百分点。

主流程把数据加载、训练、保存串起来：

import numpy as np import torch import torch.nn as nn from torch.utils.data import DataLoader, TensorDataset def main(): train_data = np.load("data/processed/train.npz") val_data = np.load("data/processed/val.npz") train_x, train_y = train_data["x"], train_data["y"] val_x, val_y = val_data["x"], val_data["y"] train_loader = DataLoader( TensorDataset(torch.from_numpy(train_x).float(), torch.from_numpy(train_y).long()), batch_size=64, shuffle=True, num_workers=4, ) val_loader = DataLoader( TensorDataset(torch.from_numpy(val_x).float(), torch.from_numpy(val_y).long()), batch_size=64, shuffle=False, num_workers=4, ) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") model = FaultCNN1D(num_classes=4).to(device) criterion = nn.CrossEntropyLoss() optimizer = torch.optim.Adam(model.parameters(), lr=1e-3) early_stopping = EarlyStopping(patience=15) best_acc = 0.0 for epoch in range(100): train_loss = train_one_epoch(model, train_loader, optimizer, criterion, device) trues, preds = evaluate(model, val_loader, device) val_acc = (np.array(trues) == np.array(preds)).mean() print(f"epoch {epoch:02d} | loss {train_loss:.4f} | val_acc {val_acc:.4f}") if val_acc > best_acc: best_acc = val_acc torch.save(model.state_dict(), "checkpoints/best.pt") if early_stopping.should_stop(val_acc): break

注意torch.from_numpy(...).long()把标签转成 long，CrossEntropyLoss 不接受 float 标签。num_workers=4在有足够内存时能明显加速数据读取，现场机器核数少就改成 1。保存用state_dict()而不是整个 model 对象，这样加载时不需要依赖原来的类定义路径，部署更干净。

checkpoint 保存最好带上验证集准确率：

torch.save({ "model": model.state_dict(), "val_acc": val_acc, "epoch": epoch, "config": cfg, }, "checkpoints/best.pt")

保存 config 的作用是后悔药。现场三个月后翻出一份模型，如果没有当时的窗口长度和采样率参数，你根本不知道自己训的是什么数据。

4.3 超参数配置化：把学习率、窗口长度写进 yaml

超参数写死在代码里是最常见的项目烂摊子。窗口长度、步长、学习率、batch size、早停 patience，这些参数现场调试时几乎都会改。写进 yaml 后，改参数不用翻代码，也不会不小心改坏逻辑。

# config.yaml data: train_path: "data/processed/train.npz" val_path: "data/processed/val.npz" sample_rate: 25600 window_size: 10240 step_size: 5120 model: num_classes: 4 base_channels: 16 train: batch_size: 64 learning_rate: 0.001 epochs: 100 early_stop_patience: 15 num_workers: 4 seed: 42 save: best_model: "checkpoints/best.pt"

train.py 开头加几行读取 yaml，后面所有参数一律从cfg取，代码里不出现裸的数字常量。seed 一定要固定，否则每次运行数据集打乱顺序不同，模型结果无法复现，也没法判断改动是好是坏。数据路径、模型保存路径也放进 yaml，换一台机器部署时，只要改文件路径，不用动代码。

5. 避坑指南：机械设备故障诊断系统最常见的 4 个坑与排查顺序

这个章节写的是我自己反复踩过的坑，也是帮别人看这类项目时最常见的共性问题。每一条按现象、原因、解决三段写，方便你在现场照着排查。

5.1 按样本随机切分，验证集虚高，现场准确率暴跌

现象：训练准确率 90%，验证准确率却能到 98%，换到另一台同型号设备测试，准确率直接掉到 70% 以下。

原因：切完窗口后直接按样本随机划分训练集和验证集，同一台设备连续采样切出的相邻窗口被同时分到两边。模型在训练时看过验证集里相似度极高的波形，本质上是把验证集的答案背下来了。

解决：按设备编号切分，确保验证集里出现的设备没有参与过训练。参考 2.3 的split_by_machine。如果现场只有一台设备，就按时间分段切，用前 70% 时间段的样本训练，用后 30% 验证，模拟“未来数据”。切分这步要在预处理阶段一次完成，后面训练和评估都用同一份划分，不要临时再切。

5.2 标签错位：模型学的不是故障，是窗口序号

现象：训练损失下降缓慢甚至不降，验证准确率长时间停在类别数分之一附近，比如 4 分类就卡在 25% 左右。

原因：窗口切好后，标签没有跟着索引对齐。常见于窗口长度和步长不一致的时候，人工给窗口标号时写错了位置；或者设备状态是按时段标记的，切窗口时窗口边界跨过了状态切换点，标签取到了错误状态。

解决：不要手动给每一段窗口标标签，用一个可视化脚本打印窗口起点和对应时间戳，检查起点所在的设备状态是否和标签一致。对齐逻辑简单验证一下：

# 打印前 5 个窗口起点和对应时刻的设备状态 for start in range(0, len(signal) - window_size + 1, step_size): if start // step_size >= 5: break timestamp = start / sampling_rate print(f"window start={start}, time={timestamp:.2f}s, state={state_at(timestamp)}")

state_at(timestamp)是从设备运行记录里查到的状态，比如 normal、inner_fault、outer_fault。如果窗口起点落在两个状态的边界上，这个窗口要么删掉，要么按起点的状态标，千万别按窗口中心标。标签错位这个问题，错一个样本影响不算大，错一批样本模型就直接学废了。

5.3 类别不均衡：故障样本太少，模型永远输出“正常”

现象：整体准确率看着有 90%，但混淆矩阵里故障类别几乎全是 0 查全率，模型把所有样本都判成正常。

原因：正常样本占 90% 以上，交叉熵损失被多数类主导，模型发现全判正常也能拿到很低的 loss，就不再学故障特征了。

解决：先用WeightedRandomSampler做过采样，原理是给故障类样本更高的采样权重，让每个 batch 里正常和故障的比例不那么悬殊。

from torch.utils.data import WeightedRandomSampler labels_np = train_y.numpy() class_counts = np.bincount(labels_np) weights = 1.0 / class_counts[labels_np] sampler = WeightedRandomSampler(weights, num_samples=len(weights), replacement=True) train_loader = DataLoader( TensorDataset(torch.from_numpy(train_x).float(), torch.from_numpy(train_y).long()), batch_size=64, sampler=sampler, )

weights按每个样本所属类别的样本数倒数构造，故障类样本少反而被多抽。replacement=True允许同一个故障样本被重复采样，缓解样本量不足。如果故障样本少到只有几十个，过采样不够用，先做数据增强：振动信号可以加小幅高斯噪声、做时间平移、把幅值缩放 0.9～1.1 倍，比在模型上折腾有效得多。

5.4 部署后准确率骤降：先查传感器，再查数据分布

现象：实验室验证集准确率 95%，装到现场只有 70%，而且误报集中在一两个类别上。

原因：现场传感器安装位置和实验室不同，比如从设备垂直方向换成了水平方向；现场转速波动、负载变化、周围其他设备振动混入，导致数据分布偏移。模型学到的是实验室数据的分布，现场波形长得不像是很正常的。

解决：先做快速分布检查，用 2.2 的extract_fft_spectrum对比现场和训练集的频谱，看峰值频率是否整体偏移。如果传感器位置换了，优先重新采集现场数据，用少量现场样本微调模型；微调时冻结前两层卷积，只更新后面几层和分类层，防止现场样本少导致过拟合。如果只是噪声水平不同，先做带通滤波，把训练时关注的频段留住，再重新评估。

6. 再往前一步：用滑窗推理与置信度阈值做现场验证

训练和评估都跑通之后，模型还只是个静态评测报告。设备现场需要的是实时判断：传感器一直传数据，系统不能每隔几分钟才出一张报告。我一般的做法是把滑动窗口搬到推理阶段，让模型对连续数据流做滚动预测，而不是等一整段离线数据。下面这段代码接收最新一段缓冲数据，输出多个窗口的平均概率和最终类别。

def online_inference(model, buffer, window_size, step_size, device): """对连续信号缓冲做滑窗推理，返回平均概率和最终类别。""" model.eval() probs = [] with torch.no_grad(): for start in range(0, len(buffer) - window_size + 1, step_size): window = buffer[start:start + window_size] x = torch.from_numpy(window).float().unsqueeze(0).unsqueeze(0).to(device) p = torch.softmax(model(x), dim=1).cpu().numpy()[0] probs.append(p) avg_prob = np.mean(probs, axis=0) cls = int(avg_prob.argmax()) return cls, avg_prob

单窗口预测很冲动，稍微一点噪声扰动就会在类别之间跳变。对多个窗口的概率取平均，等于做了一个时间上的平滑，比单独看某一次预测稳很多。step_size决定推理间隔，也决定系统多久输出一次诊断结果；窗口越长，看到的上下文越多，但延迟也越高。

现场部署时，我给每个类别设一个置信度阈值，通常 0.7～0.8。低于阈值就标成“不确定”，不触发报警；即使超过阈值，也要连续三个窗口都判同一故障才报警。这套逻辑解决的是信任问题：设备老师傅被假警报折腾几次，后面真故障也没人信了。我现在的习惯是第一版就把置信度阈值和连续报警次数写进配置文件，现场先用三到五天试跑，宁可缓报，也不要误报。跑过一轮之后，再根据误报和漏报的比例回调阈值。这套故障诊断系统的价值，其实不是模型准确率多少，而是能不能让现场人员把警报当回事。希望帮到你。

本文还有配套的精品资源，点击获取