Розробка AI-системи для моніторингу енергомереж

Розробка AI-системи для моніторингу енергомереж

Енергомережа – критична інфраструктура, де аномалія протягом секунд може стати аварією. AI-моніторинг обробляє дані з PMU (Phasor Measurement Unit) із частотою 50 Гц, дані SCADA кожні 4 секунди та сигнали релейного захисту в реальному часі.

Архітектура даних енергомережі

Джерела телеметрії:

data_streams = {
    'pmu_synchrophasors': {
        'frequency': '50 Hz (50 readings/sec)',
        'measures': ['voltage_magnitude', 'voltage_angle', 'current_magnitude',
                     'frequency', 'ROCOF'],  # Rate of Change of Frequency
        'standard': 'IEEE C37.118'
    },
    'scada_ems': {
        'frequency': '4 seconds',
        'measures': ['active_power_mw', 'reactive_power_mvar', 'transformer_load_pct',
                     'bus_voltage_kv', 'line_current_a']
    },
    'smart_meters': {
        'frequency': '15 minutes',
        'measures': ['energy_kwh', 'peak_demand', 'power_factor']
    },
    'weather': {
        'frequency': '10 minutes',
        'measures': ['temperature', 'wind_speed', 'solar_irradiance', 'humidity']
    }
}

Детекція аномалій напруги та частоти

Real-time моніторинг якості електроенергії:

import numpy as np

class PowerQualityMonitor:
    # Нормативы: ГОСТ 32144-2013 / EN 50160
    FREQ_NOMINAL = 50.0      # Гц
    FREQ_TOLERANCE = 0.2     # ±0.2 Гц в норме
    VOLTAGE_TOLERANCE = 0.10 # ±10% от номинала

    def __init__(self, nominal_voltage_kv: float):
        self.nominal_voltage = nominal_voltage_kv
        self.history = []

    def analyze_sample(self, timestamp, voltage_kv: float,
                       frequency_hz: float, current_a: float) -> dict:
        events = []

        # Отклонение частоты
        freq_deviation = abs(frequency_hz - self.FREQ_NOMINAL)
        if freq_deviation > self.FREQ_TOLERANCE:
            events.append({
                'type': 'frequency_deviation',
                'value': frequency_hz,
                'deviation': freq_deviation,
                'severity': 'critical' if freq_deviation > 0.5 else 'warning'
            })

        # Отклонение напряжения
        voltage_deviation_pct = abs(voltage_kv - self.nominal_voltage) / self.nominal_voltage
        if voltage_deviation_pct > self.VOLTAGE_TOLERANCE:
            events.append({
                'type': 'voltage_deviation',
                'value': voltage_kv,
                'deviation_pct': voltage_deviation_pct * 100,
                'direction': 'undervoltage' if voltage_kv < self.nominal_voltage else 'overvoltage',
                'severity': 'critical' if voltage_deviation_pct > 0.15 else 'warning'
            })

        # ROCOF (Rate of Change of Frequency) — предвестник нестабильности
        if len(self.history) > 0:
            rocof = (frequency_hz - self.history[-1]['frequency']) / 0.02  # Hz/s (50Hz → 20мс)
            if abs(rocof) > 1.0:  # > 1 Гц/с = значительное возмущение
                events.append({
                    'type': 'high_rocof',
                    'value': rocof,
                    'severity': 'critical' if abs(rocof) > 2.0 else 'warning'
                })

        self.history.append({'frequency': frequency_hz, 'timestamp': timestamp})
        if len(self.history) > 1000:
            self.history.pop(0)

        return {'timestamp': timestamp, 'events': events, 'healthy': len(events) == 0}

Прогнозування навантаження (Load Forecasting)

Короткостроковий прогноз для диспетчера:

from sklearn.ensemble import GradientBoostingRegressor
import pandas as pd

def build_load_forecasting_model(historical_load: pd.DataFrame) -> GradientBoostingRegressor:
    """
    Прогноз на 24-48 часов для планирования генерации и предотвращения перегрузок.
    """
    features = [
        'hour', 'day_of_week', 'month', 'is_holiday',
        'temperature', 'temperature_forecast',
        'load_1h_ago', 'load_24h_ago', 'load_168h_ago',  # лаги
        'load_trend_24h'  # slope за последние 24 часа
    ]

    historical_load['load_1h_ago'] = historical_load['load_mw'].shift(4)   # 15-мин данные
    historical_load['load_24h_ago'] = historical_load['load_mw'].shift(96)
    historical_load['load_168h_ago'] = historical_load['load_mw'].shift(672)

    model = GradientBoostingRegressor(
        n_estimators=300,
        max_depth=5,
        learning_rate=0.05
    )
    train_data = historical_load.dropna(subset=features)
    model.fit(train_data[features], train_data['load_mw'])
    return model

Детекція перевантажень та запобігання каскадним відмовам

Розрахунок ризику навантаження трансформатора:

def assess_transformer_overload_risk(transformer_data: pd.DataFrame,
                                      load_forecast: pd.Series,
                                      rated_mva: float) -> dict:
    """
    Трансформаторы допускают кратковременные перегрузки по ГОСТ 14209.
    1.3 × Sном — допустимо 2 часа при нормальной температуре.
    """
    current_load_pct = transformer_data['load_mw'].iloc[-1] / (rated_mva * 0.9) * 100

    # Тепловая модель трансформатора (упрощённая)
    ambient_temp = transformer_data['ambient_temp'].iloc[-1]
    winding_temp_est = ambient_temp + 65 * (current_load_pct / 100) ** 2  # Hotspot

    # Прогноз перегрузки
    max_forecast_load_pct = load_forecast.max() / (rated_mva * 0.9) * 100

    overload_risk = 'none'
    if max_forecast_load_pct > 130:
        overload_risk = 'critical'
    elif max_forecast_load_pct > 110:
        overload_risk = 'warning'
    elif max_forecast_load_pct > 100:
        overload_risk = 'caution'

    return {
        'current_load_pct': round(current_load_pct, 1),
        'winding_temp_est_c': round(winding_temp_est, 1),
        'max_forecast_load_pct': round(max_forecast_load_pct, 1),
        'overload_risk': overload_risk,
        'recommended_action': 'load_shedding' if overload_risk == 'critical' else None
    }

Детекція розкрадання електроенергії

Порівняння балансів:

def detect_commercial_losses(feeder_data: pd.DataFrame,
                               meter_data: pd.DataFrame) -> dict:
    """
    Коммерческие потери = технические потери + хищения.
    Аномалия: потери сегмента > ожидаемых по модели.
    """
    # Технические потери из модели ЛЭП (I²R)
    technical_losses_model = calculate_technical_losses(
        feeder_data['current_a'],
        feeder_data['resistance_ohm']
    )

    actual_losses = feeder_data['supply_mwh'].sum() - meter_data['consumed_mwh'].sum()
    commercial_losses = actual_losses - technical_losses_model

    loss_rate = commercial_losses / feeder_data['supply_mwh'].sum()

    return {
        'technical_losses_mwh': technical_losses_model,
        'commercial_losses_mwh': round(commercial_losses, 2),
        'loss_rate_pct': round(loss_rate * 100, 2),
        'anomaly': loss_rate > 0.08,  # > 8% = подозрительно
        'action': 'field_inspection' if loss_rate > 0.15 else None
    }

Інтеграція з EMS/SCADA: OSIsoft PI historian, GE Grid Solutions EMS, Siemens SICAM. Оперативний дашборд диспетчер з CIM-моделлю мережі. Алерти через SMS та систему диспетчеризації.

Терміни: Real-time моніторинг KPI + алерти напруги/частоти + load forecast - 4-5 тижнів. Теплова модель трансформаторів, детекція розкрадань, каскадний аналіз, EMS інтеграція – 3-4 місяці.