CWRU轴承故障诊断实战指南(一)：数据加载与预处理全流程解析

1. CWRU轴承数据集入门指南

第一次接触CWRU轴承数据集时，我也被它复杂的目录结构和各种参数搞得晕头转向。这个由美国凯斯西储大学发布的经典数据集，包含了从正常到不同故障状态的轴承振动信号，是机械故障诊断领域的"MNIST"。但和MNIST不同，CWRU数据集需要处理采样频率、负载工况、故障类型等多维度参数，这对新手确实不太友好。

我建议你先从12K采样频率的数据入手，这是最常用的子集。数据集主要包含两类文件：正常状态数据（Normal Baseline Data）和故障数据（12k Drive End Bearing Fault Data）。每个.mat文件都存储了特定工况下的振动信号，文件名中的数字编码了负载和故障信息。比如"097.mat"代表0负载下的正常状态，"105.mat"则是0负载下内圈0.007英寸故障的数据。

理解这些编码规则很重要，因为后续我们需要根据文件名自动分类数据。举个例子，文件名的前两位数字减去97就是负载值（0-3对应0-3马力），而特定范围的数字对应不同故障类型和程度。这个映射关系我会在代码实现部分详细说明。

2. 数据加载函数深度解析

2.1 load_data函数参数详解

load_data函数是这个项目的核心，它有五个关键参数需要理解：

• num：控制每类故障的样本数量。我建议初期设置为50-100，既能保证模型训练效果，又不会消耗太多内存。注意这个值不能超过实际数据量，函数会自动检查并提示最大值。

• length：每个样本的长度。CWRU原始信号很长，我们需要截取片段。1280是个不错的起点，对应0.1秒的振动信号（12kHz采样率）。太短会丢失特征，太长会增加计算负担。

• hp：负载工况列表。默认[0,1,2]代表0-2马力三种工况。如果想挑战更复杂的跨工况诊断，可以加入3马力数据。

• fault_diameter：故障直径列表。默认包含0.007和0.028英寸两种程度。实际还有0.014和0.021英寸的数据可供探索。

• split_rate：数据集划分比例。我习惯用[0.7,0.2,0.1]的比例分配训练集、验证集和测试集。验证集用于调参，测试集必须严格隔离。

2.2 数据加载的实现细节

函数内部的处理流程值得仔细研究。首先是路径处理部分：

bath_path1 = r"path\to\Normal Baseline Data\\" bath_path2 = r"path\to\12k Drive End Bearing Fault Data\\"

这里需要注意两点：路径中的双反斜杠是为了转义，r前缀防止转义字符被解析。我建议使用绝对路径，或者将数据集放在项目根目录下。

数据加载的核心逻辑是遍历所有指定的负载和故障类型：

for i in hp: # 加载正常数据 normal_data = open_data(bath_path1, 97+i) # 加载各类故障数据 for j in fault_diameter: inner_data = open_data(bath_path2, inner_num + i) ball_data = open_data(bath_path2, ball_num + i) outer_data = open_data(bath_path2, outer_num + i)

这里用97作为基准值是因为正常数据的文件从097.mat开始编号。故障数据的编号规则更复杂，需要根据故障直径选择不同的基准值。

3. 数据预处理关键技术

3.1 信号归一化处理

原始振动信号的幅值范围可能相差很大，直接输入模型会导致权重更新不稳定。我们使用MinMaxScaler进行归一化：

min_max_scaler = preprocessing.MinMaxScaler() data = min_max_scaler.fit_transform(np.transpose(data,[1,0])) data = np.transpose(data,[1,0])

这里有个技巧：先转置再归一化是为了对每个特征单独缩放。如果不转置，会变成对时间点的全局缩放，这会破坏信号的相对关系。

3.2 标签编码策略

CWRU数据集本质是多分类问题，我们需要将故障类型编码为one-hot向量。关键是要建立一致的标签映射：

label = 0 # 正常状态 label += 1 # 内圈故障 label += 2 # 滚珠故障 label += 3 # 外圈故障

注意标签值会随着负载和故障程度组合而变化。比如3种负载×（1正常+3故障类型×2种程度）=21类。np_utils.to_categorical会自动处理这种多分类编码。

4. 数据集划分最佳实践

4.1 随机划分与数据平衡

确保每类样本数量相同很重要，否则模型会偏向多数类。函数中这段代码实现了自动平衡：

num_list = [len(i) for i in data_list] min_num = min(num_list) if num > min_num: print("警告：样本数量超出上限，自动调整为%d" % min_num)

我建议在划分前先打乱数据顺序，避免时间相关性的影响。使用random.sample可以确保真正的随机性，而不是简单的切片。

4.2 三种数据集的正确使用

训练集用于模型参数更新，验证集用于早停和超参调优，测试集只用于最终评估。常见的错误是：

1. 用测试集做验证，导致信息泄露
2. 划分时没有保持类别比例（即分层抽样）
3. 在不同负载间简单划分，应该确保每种负载都出现在所有数据集中

我们的实现通过在每类数据内部划分来解决这些问题：

for data in data_list: data = data[0:min_num,:] a,b,c = split_data(data,split_rate) train.append(a) eval.append(b) test.append(c)

5. 完整代码实现与调试技巧

5.1 数据加载完整代码

以下是整合了所有功能的完整实现：

import random import numpy as np import scipy.io as scio from sklearn import preprocessing from keras.utils import np_utils def deal_data(data, length, label): # 数据处理实现 pass def open_data(bath_path, key_num): # 文件加载实现 pass def split_data(data, split_rate): # 数据划分实现 pass def load_data(num=90, length=1280, hp=[0,1,2], fault_diameter=[0.007,0.028], split_rate=[0.7,0.2,0.1]): # 完整加载逻辑 pass

5.2 常见问题排查

1. 文件找不到错误：检查路径是否正确，特别注意Windows和Linux的路径差异
2. 内存不足：减少num或length的值，或者使用生成器逐步加载
3. 标签不对齐：检查fault_diameter和hp的组合是否与文件名匹配
4. 归一化异常：确认转置操作是否正确，可以打印data.shape验证

我在实际项目中遇到过最棘手的问题是不同Python版本对.mat文件的解析差异。如果遇到类似问题，可以尝试：

data = scio.loadmat(path, verify_compressed_data_integrity=False)

6. 进阶应用与扩展思路

6.1 跨工况诊断挑战

尝试修改hp参数包含所有四种负载[0,1,2,3]，观察模型在不同负载间的泛化能力。这模拟了实际工业场景中设备运行条件变化的情况。可以从简单到复杂分三步实验：

1. 相同负载训练和测试
2. 不同负载但相同故障类型
3. 全新负载和故障组合

6.2 故障程度回归分析

除了分类任务，还可以尝试预测故障程度（直径）。这需要修改标签系统：

# 将分类标签改为故障直径值 label = fault_diameter[index] # 使用MSE损失代替交叉熵 model.compile(loss='mse', optimizer='adam')

6.3 时频分析特征提取

原始振动信号可以转换为时频图像作为CNN输入。常用的方法有：

1. 短时傅里叶变换(STFT)
2. 连续小波变换(CWT)
3. 希尔伯特-黄变换(HHT)

这里给出STFT的实现示例：

from scipy import signal f, t, Zxx = signal.stft(data, fs=12000, nperseg=256)

7. 工程实践建议

在实际部署这类系统时，有几点经验值得分享：

1. 数据版本控制：使用DVC或Git LFS管理数据集版本，确保实验可复现
2. 自动化测试：为数据加载编写单元测试，验证样本形状、标签分布等
3. 监控数据漂移：定期检查新采集数据的统计特性是否与训练数据一致
4. 内存优化：对于大型数据集，考虑使用HDF5格式存储预处理后的数据

一个容易忽视但重要的细节是时间戳对齐。虽然CWRU数据已经对齐，但在实际项目中，多传感器数据的时间同步往往是第一个要解决的问题。我习惯在数据加载阶段就加入时间校验：

def check_timestamps(data, expected_fs=12000): interval = np.diff(data[:,0]) assert np.allclose(interval, 1/expected_fs, rtol=0.01), "采样率异常"