光伏电站告警太多,应该如何做分级诊断?

光伏告警引擎不能单靠规则(误报多)或单靠 ML(可解释性差)。双轨是工业级方案。

一、双轨架构

graph TB
    A["设备数据"] --> B["规则引擎"]
    A --> C["ML 模型"]
    B --> D["规则告警"]
    C --> E["AI 异常"]
    D --> F["告警融合"]
    E --> F
    F --> G["分级告警"]

二、规则引擎(Python)

from dataclasses import dataclass
from typing import Callable

@dataclass
class AlertRule:
    name: str
    severity: str
    condition: Callable[[dict], bool]
    message_template: str

RULES = [
    AlertRule(
        name="inverter_offline",
        severity="P0",
        condition=lambda d: d["last_data_min_ago"] > 15,
        message_template="逆变器 {inverter_id} 离线超 15 分钟"
    ),
    AlertRule(
        name="power_too_low",
        severity="P1",
        condition=lambda d: d["power_kw"] < d["expected_power_kw"] * 0.5 and 8 <= d["hour"] <= 17,
        message_template="逆变器 {inverter_id} 白天功率异常低"
    ),
    AlertRule(
        name="high_temperature",
        severity="P1",
        condition=lambda d: d["temperature_c"] > 75,
        message_template="逆变器 {inverter_id} 温度 {temperature_c}℃ 过高"
    ),
    AlertRule(
        name="frequent_faults",
        severity="P1",
        condition=lambda d: d["fault_count_24h"] > 5,
        message_template="逆变器 {inverter_id} 24h 内故障 {fault_count_24h} 次"
    ),
]

class RuleEngine:
    def __init__(self, rules: list[AlertRule]):
        self.rules = rules

    def check(self, data: dict) -> list[dict]:
        alerts = []
        for rule in self.rules:
            if rule.condition(data):
                alerts.append({
                    "rule": rule.name,
                    "severity": rule.severity,
                    "message": rule.message_template.format(**data)
                })
        return alerts

三、ML 异常检测

import torch
import torch.nn as nn

class AnomalyAE(nn.Module):
    def __init__(self, input_dim=10):
        super().__init__()
        self.encoder = nn.Sequential(
            nn.Linear(input_dim, 16), nn.ReLU(),
            nn.Linear(16, 4),
        )
        self.decoder = nn.Sequential(
            nn.Linear(4, 16), nn.ReLU(),
            nn.Linear(16, input_dim),
        )

    def forward(self, x):
        return self.decoder(self.encoder(x))

class MLAnomalyDetector:
    def __init__(self, model_path: str):
        self.model = AnomalyAE()
        self.model.load_state_dict(torch.load(model_path))
        self.model.train(False)
        self.threshold = 0.05

    def detect(self, features: list[float]) -> tuple[bool, float]:
        x = torch.tensor(features, dtype=torch.float32).unsqueeze(0)
        with torch.no_grad():
            pred = self.model(x)
        error = ((x - pred) ** 2).mean().item()
        return error > self.threshold, error

四、告警融合

class AlertFusion:
    """规则 + ML 双轨融合"""

    def __init__(self, rule_engine, ml_detector):
        self.rule_engine = rule_engine
        self.ml_detector = ml_detector

    def process(self, data: dict) -> list[dict]:
        rule_alerts = self.rule_engine.check(data)
        ml_anomaly, ml_score = self.ml_detector.detect(self._extract_features(data))

        final_alerts = list(rule_alerts)

        if ml_anomaly:
            severity = "P0" if ml_score > 0.15 else "P1" if ml_score > 0.10 else "P2"

            # 检查是否跟规则告警重叠
            is_new = not any(a["severity"] in ["P0", "P1"] for a in rule_alerts)
            if is_new:
                final_alerts.append({
                    "rule": "ml_anomaly",
                    "severity": severity,
                    "message": f"AI 检测异常(score={ml_score:.3f}),请人工核查",
                    "ml_score": ml_score,
                })

        return final_alerts

    def _extract_features(self, data: dict) -> list[float]:
        return [
            data["power_kw"], data["voltage_v"], data["current_a"],
            data["temperature_c"], data["irradiance"],
            data["hour"], data["fault_count_24h"],
            data["power_kw"] / data["expected_power_kw"],
            data["last_data_min_ago"],
            data["theta_pr"],
        ]

五、告警降噪

class AlertDeduplication:
    """告警去重 + 抑制"""

    def __init__(self):
        self.recent_alerts: dict[str, datetime] = {}
        self.suppression_window = timedelta(minutes=30)

    def is_duplicate(self, alert: dict) -> bool:
        key = f"{alert['rule']}:{alert.get('inverter_id', '')}"
        last_time = self.recent_alerts.get(key)
        if last_time and datetime.now() - last_time < self.suppression_window:
            return True
        self.recent_alerts[key] = datetime.now()
        return False

六、告警升级

class AlertEscalation:
    """告警升级:P2 → P1 → P0 → 紧急"""

    async def escalate_if_unhandled(self, alert: dict):
        """30 分钟没处理升级"""
        await asyncio.sleep(30 * 60)
        if not await self.is_acknowledged(alert["id"]):
            if alert["severity"] == "P2":
                await self.upgrade_to("P1", alert)
            elif alert["severity"] == "P1":
                await self.upgrade_to("P0", alert)
            elif alert["severity"] == "P0":
                await self.send_emergency_call(alert)

七、推送(钉钉)

async def send_alert_to_dingtalk(alert: dict):
    webhook = get_webhook_for_severity(alert["severity"])
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"{alert['severity']} 告警",
            "text": f"### {alert['message']}\n时间:{alert['timestamp']}\n规则:{alert['rule']}"
        }
    }
    async with httpx.AsyncClient() as c:
        await c.post(webhook, json=payload)

八、ZenovaOS 视角

ZenovaOS 告警引擎用规则 + ML 双轨,实测误报率 < 5%(传统纯规则 30-40%)。

总结

光伏告警引擎用规则 + ML 双轨。规则可解释,ML 发现复杂模式,融合 + 去重 + 升级完整方案。