AI预测性维护：从传感器数据到设备故障预警的端到端方案

AI预测性维护：从传感器数据到设备故障预警的端到端方案

预测性维护的价值不在"预测"本身，而在于"在故障发生前72小时发出预警，让维修工单能在非高峰期被处理"。

一、场景与数据

2023年底我们为一家风电运维公司搭建预测性维护系统，管理约200台风机，每台风机128个传感器（振动、温度、转速、油液、电流等），采样频率1Hz~1kHz不等。核心目标：将非计划停机减少40%，将维修成本降低25%。

传感器数据的特性决定了我们的技术选型：

• 高频振动数据（1kHz~10kHz）：用于轴承/齿轮箱故障诊断，数据量大、需要频域分析
• 中频工况数据（1Hz~10Hz）：温度、转速、功率等运行参数，用于趋势预测
• 低频事件数据：告警日志、维修记录、备件更换记录

二、传感器数据的特征提取

这是整个方案最关键的环节。原始振动数据是1kHz的时间序列——每秒1000个浮点数，200台风机×128个传感器，每天产生约2TB原始数据。直接喂给模型是不可行的。

2.1 时域特征

import numpy as np
from scipy import stats
from typing import Dict
def extract_time_domain_features(signal: np.ndarray, window_size: int = 1024) -> Dict[str, float]:
    """
    从振动信号中提取时域特征
    signal: 1维振动信号数组
    window_size: 滑动窗口大小
    """
    features = {}
    # 基础统计特征
    features['mean'] = float(np.mean(signal))
    features['std'] = float(np.std(signal))
    features['rms'] = float(np.sqrt(np.mean(np.square(signal))))  # 有效值
    features['peak'] = float(np.max(np.abs(signal)))              # 峰值
    features['peak_to_peak'] = float(np.ptp(signal))              # 峰峰值
    # 峭度（对冲击信号敏感，轴承早期故障的重要指标）
    features['kurtosis'] = float(stats.kurtosis(signal))
    # 偏度（反映信号对称性）
    features['skewness'] = float(stats.skew(signal))
    # 波形因子
    features['crest_factor'] = features['peak'] / features['rms']  # 峰值因子
    features['shape_factor'] = features['rms'] / (np.mean(np.abs(signal)) + 1e-10)
    # 脉冲指标（对轴承点蚀非常敏感）
    features['impulse_factor'] = features['peak'] / (np.mean(np.abs(signal)) + 1e-10)
    # 裕度因子
    sum_sqrt = np.sum(np.sqrt(np.abs(signal)))
    features['clearance_factor'] = features['peak'] / ((sum_sqrt / len(signal)) ** 2 + 1e-10)
    return features
def extract_frequency_domain_features(signal: np.ndarray, fs: float = 1000.0) -> Dict[str, float]:
    """
    频域特征提取 - 对轴承/齿轮箱故障诊断至关重要
    """
    n = len(signal)
    fft_result = np.fft.rfft(signal)
    magnitude = np.abs(fft_result) / n
    freqs = np.fft.rfftfreq(n, 1.0 / fs)
    features = {}
    # 频谱重心
    features['spectral_centroid'] = float(
        np.sum(freqs * magnitude) / (np.sum(magnitude) + 1e-10)
    )
    # 频谱标准差
    centroid = features['spectral_centroid']
    features['spectral_spread'] = float(np.sqrt(
        np.sum(((freqs - centroid) ** 2) * magnitude) / (np.sum(magnitude) + 1e-10)
    ))
    # 频带能量分布（轴承故障频率通常集中在特定频段）
    bands = {
        'low_band': (10, 100),      # 不平衡/不对中
        'mid_band': (100, 500),     # 齿轮啮合频率
        'high_band': (500, 2000),   # 轴承共振频率
    }
    for band_name, (low, high) in bands.items():
        mask = (freqs >= low) & (freqs <= high)
        features[f'{band_name}_energy'] = float(np.sum(magnitude[mask] ** 2))
    return features

2.2 特征工程的生产化

@Service
public class FeatureExtractionPipeline {
    /**
     * Flink中调用的特征提取流程
     */
    public FeatureVector extract(SensorBatch batch) {
        double[] rawSignal = batch.getRawValues();
        // 并行提取不同类型特征
        CompletableFuture<Map<String, Double>> timeFeatures = 
            CompletableFuture.supplyAsync(() -> 
                pythonBridge.callPython("extract_time_domain_features", rawSignal));
        CompletableFuture<Map<String, Double>> freqFeatures = 
            CompletableFuture.supplyAsync(() -> 
                pythonBridge.callPython("extract_frequency_domain_features", rawSignal));
        // 合并特征向量（约40维特征）
        return timeFeatures.thenCombine(freqFeatures, (time, freq) -> {
            FeatureVector vector = new FeatureVector(batch.getSensorId(), batch.getTimestamp());
            vector.putAll(time);
            vector.putAll(freq);
            return vector;
        }).join();
    }
}

三、异常检测算法的对比与选择

我们在风机数据集上对比了三种主流异常检测算法：

最终上线方案是混合模型：

class HybridAnomalyDetector:
    """
    Isolation Forest + LSTM-AutoEncoder 混合异常检测
    IF负责快速初筛（高召回），LAE负责精细判定（高精密）
    """
    def __init__(self, contamination=0.05):
        self.if_model = IsolationForest(
            contamination=contamination,
            n_estimators=200,
            max_samples='auto',
            random_state=42
        )
        self.lae_model = self._build_lstm_autoencoder()
        self.threshold = self._calibrate_threshold()
    def _build_lstm_autoencoder(self):
        model = tf.keras.Sequential([
            # Encoder
            tf.keras.layers.LSTM(64, activation='relu', 
                return_sequences=True, input_shape=(64, 40)),
            tf.keras.layers.LSTM(32, activation='relu', return_sequences=False),
            # Bottleneck
            tf.keras.layers.RepeatVector(64),
            # Decoder
            tf.keras.layers.LSTM(32, activation='relu', return_sequences=True),
            tf.keras.layers.LSTM(64, activation='relu', return_sequences=True),
            tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(40))
        ])
        model.compile(optimizer='adam', loss='mse')
        return model
    def predict(self, features: np.ndarray) -> Tuple[bool, float, str]:
        """
        两级检测：IF初筛 → LAE确认
        Returns:
            (is_anomaly, confidence_score, detected_by)
        """
        # Level 1: IF快速初筛（宁可多报，不能漏报）
        if_score = self.if_model.score_samples(features.reshape(1, -1))[0]
        if if_score > self.if_threshold:
            return False, abs(if_score), 'IF_PASS'
        # Level 2: LAE精细判定
        reconstruction = self.lae_model.predict(features.reshape(1, 64, 40), verbose=0)
        recon_error = np.mean((features - reconstruction) ** 2)
        is_anomaly = recon_error > self.lae_threshold
        confidence = 1.0 - (recon_error / (self.lae_threshold * 3))
        return is_anomaly, min(max(confidence, 0.0), 1.0), 'LAE_CONFIRMED'

四、告警阈值的学习型调优

固定阈值是预测性维护的大忌——冬天和夏天风机的振动基线完全不同。

class AdaptiveThresholdManager:
    """
    基于运行工况的自适应阈值管理
    """
    def __init__(self, window_days=30):
        self.window_days = window_days
        self.baselines = {}  # {sensor_id: {feature: (mean, std)}}
    def update_baseline(self, sensor_id: str, features_df: pd.DataFrame):
        """
        每天凌晨用最近30天数据更新基线
        仅使用正常状态数据（排除已知告警时段）
        """
        recent = features_df[
            (features_df['timestamp'] > pd.Timestamp.now() - pd.Timedelta(days=self.window_days))
            & (features_df['is_normal'] == True)
        ]
        baselines = {}
        for col in recent.select_dtypes(include=[np.number]).columns:
            if col in ['timestamp', 'sensor_id']:
                continue
            data = recent[col].dropna()
            if len(data) > 100:
                baselines[col] = {
                    'mean': float(data.mean()),
                    'std': float(data.std()),
                    'percentile_99': float(data.quantile(0.99)),
                    'percentile_01': float(data.quantile(0.01))
                }
        self.baselines[sensor_id] = baselines
    def calculate_dynamic_threshold(self, sensor_id: str, feature_name: str, 
                                     current_value: float) -> float:
        """
        动态阈值 = 均值 ± k * 标准差，其中k根据工况自适应
        """
        baseline = self.baselines.get(sensor_id, {}).get(feature_name)
        if not baseline:
            return current_value * 0.3  # 默认30%偏差
        # 高温工况下放宽阈值（振动信号在高功率运行时本身波动大）
        k = 3.0  # 默认3σ
        if self._is_high_load_condition(sensor_id):
            k = 4.0  # 高负载放宽到4σ
        return baseline['mean'] + k * baseline['std']

五、总结

预测性维护从0到1的搭建中，我们最重要的三个认知：

1. 特征工程比模型选型更重要。我们在时域特征（峭度、脉冲因子、峰峰值）上投入的时间是模型调参的3倍，但这些特征对故障的区分度直接决定了最终效果。同一个LSTM-AutoEncoder，用原始信号训练的F1只有0.62，加上手工特征后跳到0.85。

2. 异常检测要分两级：高召回的粗筛+高精密的确认。Isolation Forest的召回率能达到95%，代价是精密率只有78%。加一层LSTM-AutoEncoder后，告警数量从每天300+降到约40条，其中真正需要处理的有35条——运维团队才愿意信任这个系统。

3. 阈值必须是自适应的。固定阈值在工况变化时要么漏报（冬天阈值对夏天太宽松），要么误报泛滥（夏天阈值对冬天太严格）。基于30天滑动窗口的动态基线，配合k值按工况自适应，是投入产出比最高的方案。

系统上线后，非计划停机减少43%，维修成本降低28%——模型预测准确率不是最终目标，设备可用率和运维成本的改善才是。