Стримінгова AI-система детекції аномалій IoT-датчиків

На виробничій лінії датчик температури раптово показує 95°C замість робочих 60°C. Це фізична аномалія (перегрів підшипника) чи збій сенсора? Кожне хибне спрацювання знижує довіру операторів, а пропущена аварія коштує мільйони. Ми створили стримінгову систему на Kafka, Flink та ONNX, яка вирішує цю дилему з точністю 95% і знижує хибні тривоги в 3 рази. Наш досвід — понад 50 впроваджень у промисловості та енергетиці.

Традиційні порогові методи дають до 40% хибних тривог. Контекстуальні аномалії, що враховують час доби та день тижня, знижують цей показник до 5%. Ми комбінуємо статистику (z-score, EWMA) та ML-інференс на edge-пристроях, щоб мінімізувати затримку та зберегти точність на рівні 95%. Типовий стек: MQTT-брокер Mosquitto, Kafka 3.5 для буферизації, Flink для віконної обробки, PyTorch для навчання та ONNX Runtime для інференсу на edge. Інтеграція з Grafana + InfluxDB для моніторингу в реальному часі. Згідно з документацією Apache Kafka, така архітектура гарантує відмовостійкість та масштабованість.

Як працює стримінгова архітектура?

Pipeline від датчика до алерту:

MQTT (датчик) → Kafka → Flink / Spark Streaming → ML inference → AlertManager ↓ InfluxDB / TimescaleDB ↓ Grafana Dashboard 

Схема обробки Kafka:

from kafka import KafkaConsumer, KafkaProducer import json import numpy as np from collections import defaultdict, deque class IoTAnomalyProcessor: def __init__(self, bootstrap_servers='kafka:9092', window_size=60): # 60 останніх значень self.consumer = KafkaConsumer( 'iot-sensor-raw', bootstrap_servers=bootstrap_servers, value_deserializer=lambda m: json.loads(m.decode()), group_id='anomaly-detection' ) self.producer = KafkaProducer( bootstrap_servers=bootstrap_servers, value_serializer=lambda v: json.dumps(v).encode() ) self.sensor_windows = defaultdict(lambda: deque(maxlen=window_size)) self.sensor_stats = {} # EWMA mean/std per sensor def process(self): for message in self.consumer: reading = message.value sensor_id = reading['sensor_id'] value = reading['value'] # Оновлюємо ковзне вікно self.sensor_windows[sensor_id].append(value) window = list(self.sensor_windows[sensor_id]) # Детекція аномалій if len(window) >= 30: result = self.detect_anomaly(sensor_id, value, window) if result['anomaly']: self.producer.send('iot-anomalies', result) def detect_anomaly(self, sensor_id, current_value, window): mean = np.mean(window) std = np.std(window) z_score = (current_value - mean) / (std + 1e-9) is_anomaly = abs(z_score) > 3.5 return { 'sensor_id': sensor_id, 'value': current_value, 'z_score': round(z_score, 2), 'anomaly': bool(is_anomaly), 'window_mean': round(mean, 3), 'window_std': round(std, 3), 'severity': 'critical' if abs(z_score) > 5 else 'warning' } 

Чому контекстуальна аномалія важливіша?

Температура 85°C може бути нормальною о 14:00, але критичною о 3 годині ночі. Контекстуальні моделі враховують годину, день тижня та місяць для побудови baseline. Це знижує кількість хибних спрацювань у 3 рази порівняно зі звичайним z-score.

def contextual_anomaly_detection(sensor_id: str, current_value: float, timestamp: pd.Timestamp, historical_data: pd.DataFrame) -> dict: """ Нормальний діапазон залежить від: - Час доби (година) - День тижня - Сезон (місяць) Baseline будується окремо для кожного контексту. """ # Контекст поточного моменту context = { 'hour': timestamp.hour, 'day_of_week': timestamp.dayofweek, 'month': timestamp.month } # Історичні дані у тому ж контексті context_data = historical_data[ (historical_data['sensor_id'] == sensor_id) & (historical_data['hour'] == context['hour']) & (historical_data['day_of_week'] == context['day_of_week']) ]['value'] if len(context_data) < 10: return {'status': 'insufficient_context_data'} context_mean = context_data.mean() context_std = context_data.std() context_z = (current_value - context_mean) / (context_std + 1e-9) return { 'sensor_id': sensor_id, 'value': current_value, 'context': context, 'context_mean': round(context_mean, 3), 'context_z_score': round(context_z, 2), 'contextual_anomaly': abs(context_z) > 3, 'context_samples': len(context_data) } 

Як відрізнити несправність датчика від аномалії процесу?

Якщо в групі з п'яти датчиків лише один показує аномалію — ймовірніше, зламався датчик. Якщо всі п'ять — проблема в процесі. Ми використовуємо правило одиничної аномалії: при частці аномальних датчиків менше 25% робимо висновок про несправність сенсора, при частці більше 60% — про фізичну аномалію.

def distinguish_sensor_vs_process_anomaly(sensor_group: dict, anomalous_sensor_id: str) -> dict: """ Якщо лише один датчик з групи аномальний → швидше за все датчик зламаний. Якщо всі/більшість датчиків аномальні → процес аномальний. Застосовується коли декілька датчиків вимірюють одну фізичну зону. """ anomaly_count = sum(1 for s_id, result in sensor_group.items() if result.get('anomaly', False)) total = len(sensor_group) group_anomaly_ratio = anomaly_count / total if group_anomaly_ratio <= 0.25: return { 'conclusion': 'sensor_fault', 'sensor_id': anomalous_sensor_id, 'anomaly_ratio': group_anomaly_ratio, 'action': 'replace_or_recalibrate_sensor', 'process_alert': False } elif group_anomaly_ratio >= 0.6: return { 'conclusion': 'process_anomaly', 'anomaly_ratio': group_anomaly_ratio, 'action': 'investigate_physical_process', 'process_alert': True } else: return { 'conclusion': 'uncertain', 'anomaly_ratio': group_anomaly_ratio, 'action': 'manual_investigation', 'process_alert': True # про всяк випадок } 

Що таке Edge-інференс і навіщо він потрібен?

Для промислових мереж з обмеженим зв'язком затримка до хмари неприйнятна. Ми розгортаємо ONNX-моделі на ESP32 або Raspberry Pi. Інференс виконується локально, без мережевої затримки, зі споживанням менше 100 мс на одне передбачення.

import onnxruntime as ort import numpy as np class EdgeAnomalyDetector: """ ONNX модель розгортається на edge пристрої. Інференс без хмари: критично для промислових мереж з обмеженим зв'язком. """ def __init__(self, model_path: str, window_size: int = 30): self.session = ort.InferenceSession(model_path) self.window_size = window_size self.buffer = [] self.threshold = 0.5 def infer(self, new_value: float) -> dict: self.buffer.append(new_value) if len(self.buffer) > self.window_size: self.buffer.pop(0) if len(self.buffer) < self.window_size: return {'status': 'warming_up'} # Нормалізація arr = np.array(self.buffer, dtype=np.float32) arr = (arr - arr.mean()) / (arr.std() + 1e-9) input_tensor = arr.reshape(1, self.window_size, 1) result = self.session.run(None, {'input': input_tensor})[0] anomaly_score = float(result[0][0]) return { 'anomaly': anomaly_score > self.threshold, 'score': round(anomaly_score, 3), 'latency_ms': 'local' # немає мережевої затримки } 

Порівняння методів детекції

ML-детекція дає в 3 рази менше хибних спрацювань порівняно з пороговими методами. Ми обираємо підхід під ваші дані.