Echtzeit-Anomalieerkennung in Python mit MQTT

Echtzeit-Anomalieerkennung in Python kann überraschend kompakt sein: ein Subscriber, ein gleitendes Fenster pro Sensor und ein Publish-Aufruf für Alarme. In diesem Tutorial bauen Sie einen Sensorsimulator und einen Streaming-Detektor, der jede MQTT-Nachricht beim Eintreffen bewertet und entprellte Anomalieereignisse veröffentlicht – alles mit dem öffentlichen MQTT-Broker, den Sie sofort nutzen können. Das Muster eignet sich als Einstieg in Predictive Maintenance für Pressen, Pumpen und Antriebe in der Fertigung.

Was Sie bauen

Der Aufbau besteht aus drei Teilen, die über Topics verbunden sind:

Ein Simulator veröffentlicht einmal pro Sekunde Temperatur- und Schwingungswerte für einige Maschinen, gelegentlich mit eingespeisten Störungen.
Ein Detektor abonniert die gesamte Telemetrie, führt gleitende Statistiken pro Maschine und Signal und berechnet für jeden Messwert einen robusten Z-Score.
Ein Alarm-Topic erhält ein JSON-Ereignis, wenn ein Signal über mehrere aufeinanderfolgende Messwerte anomal bleibt, und ein Entwarnungsereignis, wenn es zum Normalzustand zurückkehrt.

connectmq/demo/<id>/plant/<machine>/telemetry   simulator -> detector
connectmq/demo/<id>/plant/<machine>/score       detector, every reading
connectmq/demo/<id>/plant/alerts                detector, on state change

Wird für jeden Messwert ein Score, aber nur bei Zustandswechseln ein Alarm veröffentlicht, bleiben Dashboards aussagekräftig, ohne Menschen mit Benachrichtigungen zu überfluten.

Voraussetzungen

Python 3.9 oder neuer
paho-mqtt 2.x
Netzwerkzugriff auf broker.connectmq.com über Port 1883 (oder 8883 für TLS)

python -m venv venv
source venv/bin/activate        # unter Windows: venv\Scripts\activate
pip install "paho-mqtt>=2.0"

Wählen Sie eine zufällige ID für Ihr Topic-Präfix, zum Beispiel a71c2e. Der öffentliche MQTT-Broker broker.connectmq.com ist für alle lesbar und nur für Tests gedacht; verwenden Sie daher nur synthetische Daten. Der Leitfaden zum öffentlichen MQTT-Broker nennt dessen Grenzen, darunter 10 Nachrichten pro Sekunde und Client, die dieses Tutorial deutlich unterschreitet.

Schritt 1: Der Sensorsimulator

Speichern Sie dies als simulator.py. Es erzeugt realistisches Rauschen, eine langsame, tagesähnliche Drift und zufällige Störungsphasen: eine Temperaturspitze, einen Sprung in der Schwingung und einen eingefrorenen Sensor.

import json
import math
import random
import time
import uuid

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a71c2e/plant"
MACHINES = ["press1", "press2", "pump1"]

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

state = {m: {"fault": None, "left": 0, "stuck": None} for m in MACHINES}
t = 0
while True:
    for m in MACHINES:
        s = state[m]
        if s["left"] == 0 and random.random() < 0.01:
            s["fault"] = random.choice(["spike", "vibration", "stuck"])
            s["left"] = random.randint(5, 15)
        temp = 60 + 3 * math.sin(t / 300) + random.gauss(0, 0.4)
        vib = 2.0 + random.gauss(0, 0.08)
        if s["left"] > 0:
            if s["fault"] == "spike":
                temp += 8
            elif s["fault"] == "vibration":
                vib += 0.9
            elif s["fault"] == "stuck":
                s["stuck"] = s["stuck"] or round(temp, 2)
                temp = s["stuck"]
            s["left"] -= 1
            if s["left"] == 0:
                s["fault"], s["stuck"] = None, None
        payload = {"ts": time.time(), "tempC": round(temp, 2), "vibMm": round(vib, 3)}
        client.publish(f"{PREFIX}/{m}/telemetry", json.dumps(payload), qos=0)
    t += 1
    time.sleep(1)

Starten Sie das Skript und lassen Sie es laufen. Die Nachrichten können Sie mit dem Mosquitto-Client beobachten:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v

Schritt 2: Eine Bewertungsmethode für die Echtzeit-Anomalieerkennung in Python wählen

Ein einfacher Z-Score verwendet Mittelwert und Standardabweichung, die ihrerseits von genau den Ausreißern verzerrt werden, die Sie erkennen wollen. Ein robuster Z-Score nutzt stattdessen den Median und die mittlere absolute Abweichung vom Median (MAD):

robust_z = 0.6745 * (x - median) / MAD

Die Konstante skaliert den MAD so, dass der Score bei normalverteilten Daten mit einem Standard-Z-Score vergleichbar ist. Ein Schwellenwert um 3,5 ist ein gängiger Ausgangspunkt. Zwei zusätzliche Prüfungen erfassen Fehler, die der Score übersieht:

Erkennung eingefrorener Werte. Sind die letzten N Messwerte identisch, ist der Sensor wahrscheinlich eingefroren, auch wenn der Wert normal aussieht.
Aufwärmphase. Bewerten Sie erst, wenn das Fenster genügend Messwerte enthält, sonst erzeugt anfängliches Rauschen Fehlalarme.

Einen Vergleich von Z-Score, IQR und EWMA mit ihren Vor- und Nachteilen finden Sie unter Methoden zur Ausreißererkennung in Sensordaten.

Schritt 3: Der Streaming-Detektor

Speichern Sie dies als detector.py. Das Skript führt ein Fenster pro Maschine und Signal, bewertet jeden Messwert, veröffentlicht Scores und löst Alarme nach einer einfachen Persistenzregel aus: Alarm nach 3 aufeinanderfolgenden Anomalien, Entwarnung nach 5 aufeinanderfolgenden normalen Messwerten.

import json
import statistics
import time
import uuid
from collections import defaultdict, deque

import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/a71c2e/plant"
WINDOW = 120          # gespeicherte Messwerte pro Signal
WARMUP = 30           # Messwerte vor Beginn der Bewertung
THRESHOLD = 3.5       # Grenze für den robusten Z-Score
RAISE_AFTER = 3
CLEAR_AFTER = 5
STUCK_AFTER = 8

windows = defaultdict(lambda: deque(maxlen=WINDOW))
recent = defaultdict(lambda: deque(maxlen=STUCK_AFTER))
streak = defaultdict(int)
calm = defaultdict(int)
active = set()


def robust_z(window, x):
    med = statistics.median(window)
    mad = statistics.median(abs(v - med) for v in window) or 1e-6
    return 0.6745 * (x - med) / mad


def evaluate(key, x):
    w = windows[key]
    recent[key].append(x)
    stuck = len(recent[key]) == STUCK_AFTER and len(set(recent[key])) == 1
    if len(w) < WARMUP:
        w.append(x)
        return 0.0, False
    z = robust_z(w, x)
    anomalous = abs(z) > THRESHOLD or stuck
    if not anomalous:
        w.append(x)  # nur aus normalen Messwerten lernen
    return z, anomalous


def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe(f"{PREFIX}/+/telemetry", qos=0)


def on_message(client, userdata, msg):
    machine = msg.topic.split("/")[-2]
    try:
        data = json.loads(msg.payload)
    except json.JSONDecodeError:
        return
    scores = {}
    for signal in ("tempC", "vibMm"):
        if signal not in data:
            continue
        key = (machine, signal)
        z, anomalous = evaluate(key, data[signal])
        scores[signal] = round(z, 2)
        if anomalous:
            streak[key] += 1
            calm[key] = 0
        else:
            calm[key] += 1
            streak[key] = 0
        if key not in active and streak[key] >= RAISE_AFTER:
            active.add(key)
            publish_alert(client, machine, signal, "raised", z, data[signal])
        elif key in active and calm[key] >= CLEAR_AFTER:
            active.discard(key)
            publish_alert(client, machine, signal, "cleared", z, data[signal])
    client.publish(f"{PREFIX}/{machine}/score", json.dumps(scores), qos=0)


def publish_alert(client, machine, signal, state, z, value):
    event = {"ts": time.time(), "machine": machine, "signal": signal,
             "state": state, "z": round(z, 2), "value": value,
             "detector": "robust-z-v1", "threshold": THRESHOLD}
    client.publish(f"{PREFIX}/alerts", json.dumps(event), qos=1)
    print("ALERT", event)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"detector-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()

Nach etwa 30 Sekunden Aufwärmphase sollten Sie ALERT-Zeilen sehen, sobald der Simulator eine Störung einspeist, gefolgt von einem cleared-Ereignis wenige Sekunden nach ihrem Ende.

Schritt 4: Alarme beobachten

Abonnieren Sie das Alarm-Topic von einem beliebigen Ort aus:

mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v

Alarme werden mit QoS 1 veröffentlicht, damit sie einen kurzen Netzwerkaussetzer überstehen; Scores nutzen QoS 0, weil der nächste ohnehin eine Sekunde später eintrifft. Der Artikel MQTT-QoS-Stufen erklärt behandelt diesen Zielkonflikt ausführlich.

Schritt 5: Ein einfaches Dashboard ergänzen

Alarme zeigen, dass etwas passiert ist; ein Diagramm zeigt, wie es sich entwickelt hat. Da der Detektor für jeden Messwert einen Score veröffentlicht, kann jedes MQTT-fähige Dashboard Rohwerte und Scores nebeneinander darstellen, ohne den Detektor anzufassen. Einige Optionen, die gut funktionieren:

Node-RED mit den Dashboard-Nodes. Abonnieren Sie +/telemetry und +/score, trennen Sie nach Maschine und speisen Sie Liniendiagramme. Ein Switch-Node kann das Alarm-Topic in eine farbige Statusanzeige pro Maschine verwandeln.
Grafana mit einem MQTT-Datenquellen-Plugin. Nützlich, wenn Sie Grafana bereits für andere Metriken verwenden und Anomalie-Scores auf denselben Bildschirmen sehen möchten.
Eine Browserseite mit MQTT über WebSocket. Verbinden Sie sich mit MQTT.js zu wss://broker.connectmq.com:8084/mqtt und zeichnen Sie die Scores mit einer beliebigen Diagrammbibliothek.

Achten Sie beim Betrachten der Diagramme darauf, wie sich der Score kurz vor und nach einem Alarm verhält. Ein sauberer Detektor zeigt einen ruhigen Score, der während einer Störung steil ansteigt und danach wieder ruhig wird. Ein verrauschter Score, der die meiste Zeit nahe am Schwellenwert pendelt, bedeutet, dass der Schwellenwert für dieses Signal zu eng ist oder das Fenster zu kurz, um die Basislinie zuverlässig zu schätzen.

Auf einem privaten ConnectMQ-Broker liefert Ihnen der Live-Topic-Explorer denselben schnellen Überblick, ohne dass Sie etwas bauen müssen – praktisch während der Feinabstimmung.

Warum jeden Messwert bewerten und selten alarmieren

Es ist verlockend, nur Alarme zu veröffentlichen, um Bandbreite zu sparen. In der Praxis ist es jedoch der Score-Strom, mit dem Sie den Detektor später abstimmen. Sie können einen Tag mit Scores erneut abspielen, prüfen, wie viele Alarme ein anderer Schwellenwert erzeugt hätte, und Detektoren fair vergleichen. Alarme hingegen gehen an Menschen und müssen selten und aussagekräftig sein; deshalb wendet der Detektor eine Persistenzregel an, bevor er sie auslöst. Liegen beide Ausgaben auf getrennten Topics, kann jeder Konsument genau das abonnieren, was er braucht.

Checkliste zur Feinabstimmung

Parameter	Wirkung bei zu niedrigem Wert	Wirkung bei zu hohem Wert	Ausgangswert
WINDOW	Basislinie springt mit dem Rauschen	Passt sich legitimen Änderungen nur langsam an	2 bis 5 Minuten an Daten
THRESHOLD	Viele Fehlalarme	Übersieht kleine Störungen	3,5
RAISE_AFTER	Alarme bei einzelnen Störimpulsen	Verzögerte Erkennung	3 Messwerte
CLEAR_AFTER	Flatternde Alarme	Alarme bleiben zu lange offen	5 Messwerte
STUCK_AFTER	Markiert sich langsam ändernde Signale	Eingefrorene Sensoren bleiben unbemerkt	Abhängig von der Auflösung

Fehlerbehebung

Es kommen keine Nachrichten an. Prüfen Sie, ob Simulator und Detektor dasselbe Präfix verwenden und jeder Client eine eindeutige, nicht leere Client-ID hat. Zwei Clients mit derselben ID trennen sich gegenseitig immer wieder.
Dauerhafte Alarme bei einem Signal. Das Signal ist möglicherweise so stark quantisiert, dass der MAD null ist; der kleine Mindestwert lässt den Score dann explodieren. Erhöhen Sie den Mindestwert auf die Auflösung des Sensors.
Alarme werden nie zurückgesetzt. Das Fenster lernt nicht mehr, weil nach einer legitimen Niveauverschiebung jeder Wert als anomal gilt. Ergänzen Sie eine periodische Neubestimmung der Basislinie oder verwenden Sie EWMA für sich langsam ändernde Signale.
Der Detektor kommt nicht hinterher. Python verarbeitet mit diesem Muster Tausende Nachrichten pro Sekunde, doch aufwendige Modelle im Callback blockieren die Schleife. Verlagern Sie die Bewertung in eine Worker-Warteschlange oder betreiben Sie mehrere Detektoren mit Shared Subscriptions von MQTT 5.

Von der Demo zum Produktivbetrieb

Wechseln Sie zu einem privaten Broker, bevor echte Daten fließen. Stellen Sie die Verbindung auf TLS und Zugangsdaten um:

client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set()  # nutzt den CA-Speicher des Systems; ConnectMQ verwendet ein Let's-Encrypt-Zertifikat
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)

Verwenden Sie die für diese Zugangsdaten generierte Client-ID, da ConnectMQ Client-IDs an die ausgegebenen Zugangsdaten bindet. Ein verwalteter MQTT-Broker bedeutet außerdem, dass der Detektor nicht das Einzige ist, was Sie am Laufen halten müssen. Im Werk ersetzen Sie das Demo-Präfix durch die ISA-95-Hierarchie Ihres Unified Namespace (UNS), etwa acme/werk-muenchen/presswerk/linie-1/presse-1/score; so landen Scores und Alarme direkt neben den Rohdaten der Maschine, und MES oder Instandhaltungssysteme finden sie ohne zusätzliche Zuordnung. Reichen Statistiken für Einzelsignale nicht mehr aus, ist ein multivariates Modell wie Isolation Forest für Sensordaten der nächste Schritt. Das architektonische Gesamtbild finden Sie unter IoT-Anomalieerkennung: Methoden und Architektur und im Leitfaden zu KI und IoT.

Mit ConnectMQ starten

Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Tarif, ganz ohne Kreditkarte, übernehmen Sie den generierten Benutzernamen, das Passwort und die Client-ID in den Detektor und beobachten Sie Scores und Alarme im Live-Topic-Explorer.

Häufig gestellte Fragen

Warum einen robusten Z-Score statt eines normalen Z-Scores verwenden?

Median und MAD werden von den Ausreißern, die Sie erkennen wollen, kaum beeinflusst, sodass die Basislinie während einer Störung stabil bleibt. Ein normaler Mittelwert und eine Standardabweichung werden zur Anomalie hingezogen und verdecken sie.

Wie viele Nachrichten pro Sekunde verarbeitet dieser Python-Detektor?

Die gleitenden Statistiken sind günstig zu berechnen, daher begrenzen meist Netzwerk und JSON-Parsing den Durchsatz, nicht die Bewertung. Für höhere Volumina betreiben Sie mehrere Detektorinstanzen mit Shared Subscriptions auf einem Broker, der diese unterstützt.

Kann ich den Detektor auf einem Raspberry Pi ausführen?

Ja. Das Skript verwendet nur die Standardbibliothek plus paho-mqtt und läuft problemlos auf einem Raspberry Pi, der als Edge-Gateway dient.

Sollten Alarme als Retained Messages gesendet werden?

Die Alarmereignisse selbst nicht. Sollen neue Dashboards den aktuellen Zustand sehen, veröffentlichen Sie pro Maschine ein separates Retained-Status-Topic mit der Liste der aktiven Alarme.

Echtzeit-Anomalieerkennung in Python für MQTT-Sensordatenströme