AI Power Grid Monitoring System Development

Development of an AI system for monitoring power grids

The power grid is critical infrastructure, where an anomaly can become a disaster in seconds. AI monitoring processes data from the PMU (Phasor Measurement Unit) at a frequency of 50 Hz, SCADA data every 4 seconds, and relay protection signals in real time.

Power Grid Data Architecture

Telemetry sources:

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']
    }
}

Detection of voltage and frequency anomalies

Real-time power quality monitoring:

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

Short-term forecast for the dispatcher:

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

Overload detection and cascading failure prevention

Calculation of the risk of transformer overload:

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
    }

Electricity theft detection

Balance Comparison:

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
    }

Integration with EMS/SCADA: OSIsoft PI historian, GE Grid Solutions EMS, Siemens SICAM. Operational dispatcher dashboard with CIM network model. Alerts via SMS and dispatch system.

Deadlines: Real-time KPI monitoring + voltage/frequency alerts + load forecast — 4-5 weeks. Transformer thermal modeling, theft detection, cascade analysis, EMS integration — 3-4 months.