某工厂设备故障预警系统用时间序列分析提前3小时检测到异常振动模式工程师分享工业过程识别实战经验教你用TS数据做设备预测性维护
说实话，我刚转行做预测性维护的时候，完全被设备振动的数据迷住了。你以为那些波形图只是枯燥的数字吗？不，它们就像设备的”心电图”，每一个异常波动都可能在暗示即将发生的故障。今天我想和你聊聊，我是怎么通过时间序列分析，在某个工厂里把设备故障预警时间提前到3小时以上的。
振动数据的”故事”——从噪音里听出异常
先说说背景。那是一家汽车零部件制造厂，有几十台CNC机床、冲压机和传送带系统。老板之前最怕的就是设备突然罢工，一次停机损失几万块不说，还可能影响整条产线的交付。他们尝试过定期更换零件，但要么换早了浪费钱，要么换晚了已经出故障。
我们决定用振动传感器来监控设备状态。振动信号是最直接的——轴承磨损、齿轮松动、转子不平衡，这些都会让振动模式发生变化。问题是怎么从海量的振动数据里，找到那些”不寻常”的信号。
我常用的方法是特征提取加上异常检测。振动数据通常是10kHz以上的采样率，原始数据量巨大，直接分析不现实。我们会先提取一些关键特征，比如：
- RMS（均方根值）：反映振动的整体能量水平
- 峰值因子：脉冲型故障的敏感指标
- 峭度（Kurtosis）：对冲击信号非常敏感
- 频域特征：通过FFT转换到频域，看特征频率的变化
这些特征就像是给设备做体检时量的各项指标，单独看可能没什么问题，但放在一起就能发现异常。
时间序列分析的核心——捕捉”趋势”和”周期性”
振动数据本质上是时间序列，所以时间序列分析的方法在这里特别管用。我主要用了以下几种方法：
1. 滑动窗口统计
这是最简单但也最有效的方法之一。我们设定一个窗口大小（比如5分钟的数据），在这个窗口内计算特征的均值、标准差等统计量。然后和之前的基线做对比。
import numpy as np
import pandas as pd
from scipy import stats
import matplotlib.pyplot as plt
class VibrationAnalyzer:
def __init__(self, window_size=300, threshold=2.5):
"""
振动异常检测器
window_size: 滑动窗口大小（秒）
threshold: 异常检测阈值（标准差倍数）
"""
self.window_size = window_size
self.threshold = threshold
self.baseline = None
def extract_features(self, vibration_data):
"""
从振动信号中提取特征
vibration_data: 一维振动信号数组
"""
features = {}
# 时域特征
features['rms'] = np.sqrt(np.mean(vibration_data**2))
features['peak'] = np.max(np.abs(vibration_data))
features['crest_factor'] = features['peak'] / features['rms']
features['kurtosis'] = stats.kurtosis(vibration_data)
features['skewness'] = stats.skew(vibration_data)
# 简单频域特征（通过FFT）
fft_result = np.fft.fft(vibration_data)
freqs = np.fft.fftfreq(len(vibration_data), d=1/10000) # 假设采样率10kHz
power_spectrum = np.abs(fft_result) ** 2
# 主要频率成分的能量
features['dominant_freq'] = freqs[np.argmax(power_spectrum[:len(power_spectrum)//2])]
features['band_energy'] = {
'low': np.sum(power_spectrum[1:100]),
'mid': np.sum(power_spectrum[100:500]),
'high': np.sum(power_spectrum[500:1000])
}
return features
def update_baseline(self, features_list):
"""
建立正常状态的基线
"""
df = pd.DataFrame(features_list)
self.baseline = {
'mean': df.mean(),
'std': df.std(),
'normal_range': {}
}
for col in df.columns:
if col != 'dominant_freq' and 'band_energy' not in col:
self.baseline['normal_range'][col] = (
self.baseline['mean'][col] - self.threshold * self.baseline['std'][col],
self.baseline['mean'][col] + self.threshold * self.baseline['std'][col]
)
def detect_anomaly(self, features):
"""
检测异常
"""
anomalies = []
scores = {}
for key, value in features.items():
if key == 'dominant_freq' or 'band_energy' in key:
continue
if self.baseline and key in self.baseline['normal_range']:
lower, upper = self.baseline['normal_range'][key]
if value < lower or value > upper:
deviation = abs(value - self.baseline['mean'][key]) / self.baseline['std'][key]
anomalies.append(key)
scores[key] = deviation
# 计算综合异常分数
total_score = sum(scores.values()) / max(len(scores), 1)
return {
'is_anomaly': len(anomalies) > 0,
'anomaly_features': anomalies,
'anomaly_score': total_score,
'details': scores
}
2. 基于统计过程的检测
我们不只是看当前值，还会观察特征的演化趋势。比如RMS值可能会缓慢上升，这说明轴承可能在逐渐磨损。单纯看某一个时刻的数据可能没问题，但如果连续几个窗口都显示上升趋势，那就需要警惕了。
class TrendDetector:
"""
趋势检测器——捕捉特征的渐变趋势
"""
def __init__(self, lookback=20, trend_threshold=0.01):
self.lookback = lookback
self.trend_threshold = trend_threshold
self.history = {}
def update(self, feature_name, value):
if feature_name not in self.history:
self.history[feature_name] = []
self.history[feature_name].append(value)
# 保持固定长度的历史记录
if len(self.history[feature_name]) > self.lookback:
self.history[feature_name] = self.history[feature_name][-self.lookback:]
def check_trend(self, feature_name):
if feature_name not in self.history or len(self.history[feature_name]) < 10:
return {'trend': 'insufficient_data', 'slope': 0}
values = self.history[feature_name]
x = np.arange(len(values))
# 线性拟合计算趋势斜率
slope, _, _, _, _ = stats.linregress(x, values)
# 判断趋势是否显著
if abs(slope) > self.trend_threshold:
trend_direction = 'increasing' if slope > 0 else 'decreasing'
return {
'trend': trend_direction,
'slope': slope,
'confidence': min(abs(slope) / (self.trend_threshold * 3), 1.0)
}
else:
return {'trend': 'stable', 'slope': slope}
3. 频域分析——找到”病根”
振动信号有时域和频域两个视角。时域看整体能量变化，频域能帮我们定位具体是哪里出了问题。比如：
- 轴承故障频率：不同的轴承缺陷（内圈、外圈、滚动体）有特定的特征频率
- 齿轮故障频率：啮合频率及其谐波
- 不平衡：1X转频分量增大
- 不对中：2X转频分量增大
def analyze_frequency_domain(vibration_signal, sampling_rate=10000):
"""
频域分析——识别故障特征频率
"""
# 计算FFT
n = len(vibration_signal)
fft_result = np.fft.rfft(vibration_signal)
freqs = np.fft.rfftfreq(n, d=1/sampling_rate)
amplitude = np.abs(fft_result) / n
# 找到主要频率成分
top_n = 10
top_indices = np.argsort(amplitude)[-top_n:][::-1]
dominant_frequencies = []
for idx in top_indices:
if freqs[idx] > 10: # 过滤掉直流分量和极低频噪声
dominant_frequencies.append({
'frequency': float(freqs[idx]),
'amplitude': float(amplitude[idx]),
'relative_energy': float(amplitude[idx] ** 2 / np.sum(amplitude ** 2))
})
return {
'sampling_rate': sampling_rate,
'dominant_frequencies': dominant_frequencies,
'total_spectrum': {
'frequencies': freqs.tolist(),
'amplitudes': amplitude.tolist()
}
}
从”异常”到”预警”——时间提前量的关键
识别异常只是第一步，更重要的是要能提前预警。我们工厂的案例中，平均能在故障发生前3-6小时检测到异常信号。这个提前量是怎么来的？
关键是对”退化过程”的理解。大多数机械故障不是一夜之间发生的，而是一个渐进的过程：
正常状态 → 初期损伤 → 损伤扩展 → 性能下降 → 功能故障
我们监控的正是从”正常”到”初期损伤”这个阶段的转变。这个阶段可能持续几小时到几天，取决于故障类型和工况。
多特征融合判定
单一的异常指标可能会误报，所以我们采用多特征融合的方式提高可靠性：
class AnomalyFusion:
"""
多特征融合异常检测——降低误报率
"""
def __init__(self,
rms_weight=0.25,
kurtosis_weight=0.25,
trend_weight=0.25,
freq_weight=0.25):
self.weights = {
'rms': rms_weight,
'kurtosis': kurtosis_weight,
'trend': trend_weight,
'freq': freq_weight
}
self.alert_level = 'normal' # normal, warning, alert
def evaluate(self,检测结果, trend_results, freq_analysis):
"""
综合评估异常程度
"""
scores = {}
# RMS异常评分
rms_score = 检测结果['details'].get('rms', 0)
scores['rms'] = min(rms_score / 5.0, 1.0) # 归一化到0-1
# 峭度异常评分
kurtosis_score = 检测结果['details'].get('kurtosis', 0)
scores['kurtosis'] = min(kurtosis_score / 5.0, 1.0)
# 趋势评分
trend_score = 0
increasing_count = 0
for feat in ['rms', 'crest_factor', 'kurtosis']:
trend = trend_results.get(feat, {})
if trend.get('trend') == 'increasing':
increasing_count += 1
trend_score += trend.get('confidence', 0)
trend_score /= max(increasing_count, 1)
scores['trend'] = trend_score
# 频域评分
freq_score = 0
if freq_analysis['dominant_frequencies']:
# 检查是否有新的频率成分出现
new_frequencies = [f for f in freq_analysis['dominant_frequencies']
if f['relative_energy'] > 0.05]
freq_score = min(len(new_frequencies) * 0.3, 1.0)
scores['freq'] = freq_score
# 加权综合评分
total_score = sum(scores[k] * self.weights[k] for k in self.weights)
# 判定预警等级
if total_score > 0.7:
self.alert_level = 'alert'
elif total_score > 0.4:
self.alert_level = 'warning'
else:
self.alert_level = 'normal'
return {
'alert_level': self.alert_level,
'total_score': total_score,
'component_scores': scores,
'detail': scores
}
实战经验——那些年踩过的坑
说实话，这条路不是一帆风顺的。分享几个我踩过的坑，希望能帮你少走弯路：
坑1：把噪声当成异常
刚开始的时候，系统经常误报。后来发现，很多”异常”其实是环境噪声——冲压机的振动会影响附近的传感器，温度变化也会让传感器产生漂移。
解决方案：
- 做环境基线测试，了解正常工况下的噪声水平
- 使用带通滤波器，只关注感兴趣的频带
- 对传感器做定期校准
坑2：过度依赖单一特征
曾经有一次，RMS值一直正常，但峭度值在缓慢上升，后来才发现是轴承出现了早期点蚀。如果只看RMS，就会错过这个信号。
解决方案：
- 始终监控多个特征
- 了解不同故障类型对应的特征变化模式
坑3：忽视工况变化
设备在高速运行时和低速运行时，振动特征完全不同。如果把不同工况的数据混在一起分析，基线就会很混乱。
解决方案：
- 按工况分别建立基线
- 记录设备的运行状态（转速、负载等）作为上下文信息
class工况AdaptiveMonitor:
"""
工况自适应监控——根据不同运行状态建立基线
"""
def __init__(self):
self.baselines = {}
self.thresholds = {}
def update_baseline(self, 工况, features_list):
"""
为特定工况建立基线
"""
df = pd.DataFrame(features_list)
self.baselines[工况] = {
'mean': df.mean(),
'std': df.std(),
'samples': len(df)
}
# 设置阈值（基于历史数据的3个标准差）
self.thresholds[工况] = {}
for col in df.columns:
mean = self.baselines[工况]['mean'][col]
std = self.baselines[工况]['std'][col]
self.thresholds[工况][col] = {
'lower': mean - 3 * std,
'upper': mean + 3 * std
}
def get_anomaly_score(self, 工况, features):
"""
针对特定工况检测异常
"""
if 工况 not in self.baselines:
return {'score': 0, 'is_anomaly': False, 'note': 'no_baseline'}
scores = {}
for key, value in features.items():
if key in self.thresholds[工况]:
lower = self.thresholds[工况][key]['lower']
upper = self.thresholds[工况][key]['upper']
std = self.baselines[工况]['std'][key]
# 计算偏离程度
if value < lower:
scores[key] = (lower - value) / std
elif value > upper:
scores[key] = (value - upper) / std
else:
scores[key] = 0
total_score = np.mean(list(scores.values())) if scores else 0
is_anomaly = total_score > 2.0 # 综合阈值
return {
'score': total_score,
'is_anomaly': is_anomaly,
'component_scores': scores
}
坑4：忽视数据的可视化
有时候数值上的异常不明显，但放到图上看，趋势就很清晰了。我们后来加了实时监控大屏，可以让运维人员直观地看到设备状态。
从实验室到现场——部署时的注意事项
把模型从测试环境部署到实际工厂，和实验室里完全是两回事。几个关键经验：
数据质量比算法更重要——传感器安装位置、采样频率、信号传输的稳定性，这些都会影响最终效果
人机协同是关键——系统预警后，需要经验丰富的工程师到现场确认。我们设计了一个”确认-反馈”机制，让工程师的确认结果用来优化模型
逐步推进——不要试图一次性监控所有设备。先选几台关键设备做试点，验证效果后再推广
报警疲劳——如果误报太多，运维人员就会忽视报警。我们的经验是宁可误报几次，也不能漏掉真正的故障，但要通过多特征融合来降低误报率
这套系统的效果
经过几个月的运行，这个预测性维护系统达到了以下效果：
- 故障预警平均提前3-6小时
- 非计划停机时间减少了约70%
- 备件库存成本降低了约30%（因为可以更精准地安排更换时机）
- 单次预警确认后，真正故障的准确率达到了85%以上
当然，不完美。有些故障仍然发生得很突然，没有明显的预兆。但总体来说，这套系统给工厂带来的价值是显著的。
给你的建议
如果你也想在自己的工作中应用时间序列分析来做预测性维护，我的建议是：
从简单开始——先监控RMS和峰值因子这两个最直观的特征，看看能不能发现异常模式
理解你的设备——不同的设备有不同的故障模式，了解设备的结构和工作原理，才能知道该关注哪些特征
积累历史数据——基线数据越多越准确，异常检测的效果越好。不要指望有数据就能立竿见影
保持耐心——预测性维护系统的优化是一个持续的过程，需要不断调整参数、完善模型
时间序列分析在工业预测性维护中的应用，核心在于”理解数据的语言”。振动信号不会说谎，它们忠实地记录了设备的每一次”心跳”。我们需要做的，就是学会听懂这些话，在设备”生病”之前就发现它。
希望这些经验对你有用。如果你在实践中遇到问题，欢迎随时交流。