智能运维在能源行业的应用：风电场SCADA数据的时序异常检测与预测性维护实践

智能运维在能源行业的应用：风电场SCADA数据的时序异常检测与预测性维护实践

一、背景与问题

某风力发电企业运营着3个风电场，共156台风机（单机容量2.5MW-4.0MW，总装机容量468MW）。每台风机通过SCADA系统每秒采集80+维度的传感器数据——包括风速、转速、桨距角、齿轮箱温度、发电机轴承振动、变流器电流等。日累计数据量达1.2TB。

到2025年初，该企业面临三个运维痛点：

二、时序异常检测方案设计

2.1 整体架构：从数据采集到预测性决策

2.2 三类模型的互补策略

三、核心算法实现

#!/usr/bin/env python3
"""风电机组多维时序异常检测融合引擎"""

import numpy as np
import logging
from typing import Optional
from dataclasses import dataclass, field
import json
from sklearn.ensemble import IsolationForest
from scipy import stats

logger = logging.getLogger("wind_turbine_anomaly")

@dataclass
class TurbineSample:
"""单台风机的一个采样时刻的多维传感器数据"""
turbine_id: str
timestamp: int
wind_speed: float # 风速 (m/s)
rotor_speed: float # 转速 (rpm)
pitch_angle: float # 桨距角 (度)
gearbox_temp: float # 齿轮箱温度 (°C)
bearing_vib_x: float # 轴承振动X方向 (mm/s)
bearing_vib_y: float # 轴承振动Y方向 (mm/s)
generator_temp: float # 发电机温度 (°C)
power_output: float # 发电功率 (kW)
ambient_temp: float # 环境温度 (°C)

class WindTurbineDetector:
"""风电机组多模型融合异常检测器

融合三类检测算法：Isolation Forest（离群）、趋势检验（缓变）、
动态阈值（工况自适应），加权评分后进行决策。
"""

# 工况相关的动态阈值表：
# (风速范围下限, 风速范围上限) → (齿轮箱温度上限, 轴承振动上限)
WIND_SPEED_THRESHOLDS = {
(0, 5): (75.0, 2.5), # 低风速：温度低，振动轻
(5, 10): (82.0, 4.0), # 中风速：正常工况
(10, 15): (88.0, 5.5), # 高风速：允许较高温度和振动
(15, 25): (95.0, 7.0), # 满发风速：极限工况允许更高阈值
}

def __init__(self, contamination: float = 0.05):
self.if_model = IsolationForest(
contamination=contamination,
random_state=42,
n_jobs=-1,
)
self.is_trained = False
# 历史正常数据基线（用于趋势对比）
self.baseline: dict[str, dict[str, float]] = {}

def train_iforest(self, samples: list[TurbineSample]):
"""使用历史正常工况数据训练Isolation Forest模型"""
try:
features = []
for s in samples:
features.append([
s.wind_speed, s.rotor_speed, s.pitch_angle,
s.gearbox_temp, s.bearing_vib_x, s.bearing_vib_y,
s.generator_temp, s.power_output, s.ambient_temp,
])
X = np.array(features, dtype=np.float64)

if len(X) < 100:
logger.warning("训练数据不足（<100条），跳过IF训练")
return False

# 标准化处理
X = (X – X.mean(axis=0)) / (X.std(axis=0) + 1e-8)
self.if_model.fit(X)
self.is_trained = True
logger.info(f"Isolation Forest训练完成: samples={len(samples)}")
return True
except Exception as e:
logger.error(f"IF模型训练失败: {e}")
self.is_trained = False
return False

def detect(self, sample: TurbineSample,
history: list[TurbineSample]) -> dict:
"""
对单个样本进行多模型融合异常检测
返回: {"score": 融合评分, "alerts": 告警列表, "details": 各模型详情}
"""
try:
results = {
"score": 0.0,
"alerts": [],
"details": {
"dynamic_threshold": {},
"isolation_forest": {},
"trend_check": {},
},
}

# 模型1：工况自适应动态阈值检测（权重0.3）
dt_score, dt_alerts = self._dynamic_threshold_check(sample)
results["details"]["dynamic_threshold"] = {
"score": dt_score, "alerts": dt_alerts,
}
results["score"] += dt_score * 0.3
results["alerts"].extend(dt_alerts)

# 模型2：Isolation Forest离群检测（权重0.4）
if self.is_trained:
if_score, if_alerts = self._isolation_forest_check(sample)
results["details"]["isolation_forest"] = {
"score": if_score, "alerts": if_alerts,
}
results["score"] += if_score * 0.4
results["alerts"].extend(if_alerts)
else:
results["details"]["isolation_forest"]["warning"] = "模型未训练"

# 模型3：Mann-Kendall趋势检验（权重0.3）
if len(history) >= 20:
trend_score, trend_alerts = self._trend_check(history)
results["details"]["trend_check"] = {
"score": trend_score, "alerts": trend_alerts,
}
results["score"] += trend_score * 0.3
results["alerts"].extend(trend_alerts)
else:
results["details"]["trend_check"]["warning"] = "历史数据不足"

results["score"] = round(min(results["score"], 1.0), 4)

# 决策分级
if results["score"] > 0.75:
results["level"] = "CRITICAL"
elif results["score"] > 0.50:
results["level"] = "WARNING"
else:
results["level"] = "NORMAL"

return results

except Exception as e:
logger.error(f"异常检测失败: turbine={sample.turbine_id}, {e}")
return {"score": 0.0, "alerts": [f"检测异常: {str(e)}"], "level": "ERROR"}

def _dynamic_threshold_check(self, sample: TurbineSample) -> tuple[float, list]:
"""工况自适应动态阈值检测"""
alerts = []
score = 0.0

# 根据当前风速查找对应的阈值范围
threshold = None
for (ws_low, ws_high), thresh in self.WIND_SPEED_THRESHOLDS.items():
if ws_low <= sample.wind_speed < ws_high:
threshold = thresh
break

if threshold is None:
return 0.0, []

gearbox_limit, vib_limit = threshold

# 齿轮箱温度超标检测
if sample.gearbox_temp > gearbox_limit:
exceed_ratio = (sample.gearbox_temp – gearbox_limit) / gearbox_limit
score = max(score, min(exceed_ratio * 2, 1.0))
alerts.append(
f"齿轮箱温度超标: {sample.gearbox_temp:.1f}°C > {gearbox_limit}°C"
)

# 轴承振动超标检测
max_vib = max(sample.bearing_vib_x, sample.bearing_vib_y)
if max_vib > vib_limit:
exceed_ratio = (max_vib – vib_limit) / vib_limit
score = max(score, min(exceed_ratio * 3, 1.0))
alerts.append(f"轴承振动超标: {max_vib:.2f}mm/s > {vib_limit}mm/s")

return round(score, 4), alerts

def _isolation_forest_check(self, sample: TurbineSample) -> tuple[float, list]:
"""Isolation Forest异常检测"""
features = np.array([[
sample.wind_speed, sample.rotor_speed, sample.pitch_angle,
sample.gearbox_temp, sample.bearing_vib_x, sample.bearing_vib_y,
sample.generator_temp, sample.power_output, sample.ambient_temp,
]], dtype=np.float64)

# 标准化
features = (features – features.mean()) / (features.std() + 1e-8)

try:
# predict返回 -1（异常）或 1（正常）
prediction = self.if_model.predict(features)[0]
decision_score = self.if_model.decision_function(features)[0]

if prediction == -1:
# 归一化异常分数到0-1区间
score = min(abs(decision_score) / 0.5, 1.0)
return round(score, 4), ["Isolation Forest检测到离群样本"]
else:
return 0.0, []
except Exception as e:
logger.error(f"IF检测失败: {e}")
return 0.0, []

def _trend_check(self, history: list[TurbineSample]) -> tuple[float, list]:
"""Mann-Kendall缓变趋势检验（检测齿轮箱温度的缓升趋势）"""
try:
temps = [s.gearbox_temp for s in history[-30:]] # 取最近30个点

if len(temps) < 10:
return 0.0, []

# Mann-Kendall趋势检验
result = stats.kendalltau(range(len(temps)), temps)

trend_score = 0.0
alerts = []

# p-value < 0.05 表示存在显著趋势
if result.pvalue < 0.05:
# 计算单位时间内温度上升速率
temp_rise_rate = (temps[-1] – temps[0]) / len(temps)

if temp_rise_rate > 0.02: # 每个采样点上升超过0.02°C
trend_score = min(temp_rise_rate * 50, 1.0)
alerts.append(
f"检测到齿轮箱温度持续上升趋势: "
f"速率={temp_rise_rate:.4f}°C/sample, "
f"p={result.pvalue:.4f}"
)

return round(trend_score, 4), alerts
except Exception as e:
logger.error(f"趋势检验失败: {e}")
return 0.0, []

四、生产落地效果

五、总结

风电场SCADA数据的智能运维，核心挑战不是模型的复杂度，而是工业场景的物理约束如何与算法融合。三个关键实践：

• 工况自适应阈值 > 固定阈值：风机在不同风速下的正常温度范围差异可达20°C，用单一阈值告警必然是灾难。将风速分档映射不同的温度/振动阈值，误报率直接下降42%
• 多模型融合 > 单模型：没有一种算法能覆盖所有故障模式。Isolation Forest擅长多维关联异常，Mann-Kendall擅长缓变趋势，动态阈值擅长即时越界——三种模型的融合评分比任何单一模型都更稳定
• 异常检测 ≠ 根因诊断：本阶段实现的是"提前预警"，真正的根因定位需要结合故障知识图谱和维修记录——让预警告诉运维人员去哪里检查，而不是试图替代运维人员的专业判断

能源行业的AIOps，模型的可解释性比精度更重要。运维人员需要知道"为什么报警"，而不仅仅是"是否报警"。