Isolation Forest für Sensordaten: offline trainieren, Live-MQTT-Nachrichten bewerten

Isolation Forest auf Sensordaten ist einer der praktischsten Wege, Anomalien zu erkennen, die sich in der Kombination mehrerer Signale verbergen. In diesem Tutorial erzeugen Sie historische Daten, bilden Fenstermerkmale, trainieren ein Modell mit scikit-learn und betreiben es in einem Python-Dienst, der Live-MQTT-Nachrichten bewertet und Anomalieereignisse veröffentlicht.

Warum Isolation Forest für Sensordaten gut funktioniert

Isolation Forest beruht auf einer einfachen Idee: Anomalien sind selten und andersartig, daher lassen sie sich leichter isolieren. Der Algorithmus baut viele Zufallsbäume; jeder Baum wählt wiederholt ein zufälliges Merkmal und einen zufälligen Trennwert. Normale Punkte liegen in dichten Regionen und benötigen viele Teilungen, bis sie getrennt sind. Anomale Punkte sind schon nach wenigen Teilungen isoliert. Die durchschnittliche Pfadlänge über alle Bäume wird zum Anomalie-Score.

Dieser Aufbau passt aus mehreren Gründen gut zu Industriesensoren:

• Keine Labels nötig. Sie trainieren auf überwiegend normalem Betrieb, von dem jedes Werk reichlich hat.
• Von Natur aus multivariat. Hoher Strom ist normal, geringer Durchfluss ist normal, aber hoher Strom bei geringem Durchfluss wird schnell isoliert.
• Schnell. Das Training auf Hunderttausenden Zeilen dauert Sekunden, und die Bewertung ist günstig genug für ein Edge-Gateway.
• Wenige Hyperparameter. Meist stimmen Sie nur die Anzahl der Bäume, die Stichprobengröße und die erwartete Kontamination ab.

Es gibt auch Grenzen. Das Verfahren versteht Zeit nicht von sich aus, daher müssen Sie zeitliches Verhalten in Merkmalen abbilden. Außerdem bewertet es relativ zu den Trainingsdaten: Ein im Sommer trainiertes Modell kann den Winter als anomal einstufen.

Isolation Forest im Vergleich zu anderen Detektoren

Wenn Sie Statistiken für Einzelsignale noch nicht ausprobiert haben, lesen Sie zuerst Ausreißererkennung in Sensordaten mit Z-Score, IQR und EWMA. Isolation Forest lohnt sich, wenn diese Methoden Beziehungen zwischen Signalen übersehen.

Voraussetzungen

pip install "paho-mqtt>=2.0" scikit-learn pandas numpy joblib

Wählen Sie ein zufälliges Präfix wie connectmq/demo/if-5d21/.... Die Beispiele nutzen den öffentlichen MQTT-Broker broker.connectmq.com, auf dem alle Topics für andere lesbar sind. Er ist für Tests und Prototypen gedacht, nicht für Produktionsdaten; verwenden Sie daher nur synthetische Daten.

Schritt 1: Historische Sensordaten erzeugen

In einem echten Projekt würden Sie die Historie aus Ihrer Datenbank exportieren. Damit das Tutorial in sich geschlossen bleibt, simuliert dieses Skript eine Pumpe mit drei korrelierten Signalen: Motorstrom, Durchfluss und Ausgangsdruck.

# make_history.py
import numpy as np
import pandas as pd

rng = np.random.default_rng(42)
n = 20_000  # ein Messwert pro Sekunde, etwa 5,5 Stunden
load = 0.6 + 0.2 * np.sin(np.arange(n) / 900) + rng.normal(0, 0.02, n)

df = pd.DataFrame({
    "currentA": 10 + 12 * load + rng.normal(0, 0.3, n),
    "flowLpm": 30 + 40 * load + rng.normal(0, 1.0, n),
    "pressureBar": 2 + 3 * load + rng.normal(0, 0.08, n),
})
df.to_csv("pump_history.csv", index=False)
print(df.describe())

Alle drei Signale bewegen sich gemeinsam mit der Last. Genau diese Korrelation kann ein Schwellenwert pro Signal nicht erkennen.

Schritt 2: Fenstermerkmale bilden

Die Bewertung einzelner Messwerte ignoriert Rauschen und kurze Dynamiken. Berechnen Sie stattdessen Merkmale über ein kurzes gleitendes Fenster: den Mittelwert jedes Signals, seine Standardabweichung und einige physikalisch sinnvolle Verhältnisse.

# features.py
import numpy as np
import pandas as pd

SIGNALS = ["currentA", "flowLpm", "pressureBar"]
WINDOW = 10  # Messwerte


def window_features(frame: pd.DataFrame) -> dict:
    feats = {}
    for s in SIGNALS:
        feats[f"{s}_mean"] = frame[s].mean()
        feats[f"{s}_std"] = frame[s].std(ddof=0)
    feats["flow_per_amp"] = feats["flowLpm_mean"] / max(feats["currentA_mean"], 1e-6)
    feats["pressure_per_flow"] = feats["pressureBar_mean"] / max(feats["flowLpm_mean"], 1e-6)
    return feats


def build_feature_table(df: pd.DataFrame, step: int = 5) -> pd.DataFrame:
    rows = [window_features(df.iloc[i:i + WINDOW])
            for i in range(0, len(df) - WINDOW, step)]
    return pd.DataFrame(rows)

Die Verhältnisse sind wichtiger, als sie aussehen. Ein blockiertes Laufrad senkt den Durchfluss pro Ampere, und eine Leckage senkt den Druck pro Durchfluss – selbst wenn jeder Rohwert in seinem normalen Bereich liegt.

Schritt 3: Den Isolation Forest trainieren

# train.py
import joblib
import pandas as pd
from sklearn.ensemble import IsolationForest

from features import build_feature_table

history = pd.read_csv("pump_history.csv")
X = build_feature_table(history)

model = IsolationForest(
    n_estimators=200,
    max_samples=256,
    contamination=0.005,   # erwarteter Anteil an Anomalien in den Trainingsdaten
    random_state=0,
)
model.fit(X)

scores = -model.score_samples(X)        # höher = anomaler
threshold = float(pd.Series(scores).quantile(0.995))
joblib.dump({"model": model, "columns": list(X.columns),
             "threshold": threshold, "version": "iforest-pump-v1"},
            "iforest_pump.joblib")
print(f"trained on {len(X)} windows, threshold={threshold:.3f}")

Zwei Details verdienen Aufmerksamkeit. Erstens liefert score_samples in scikit-learn für auffälligere Punkte niedrigere Werte; durch die Negation erhalten Sie einen intuitiven Score nach dem Prinzip „höher ist schlechter“. Zweitens sollten Sie sich nicht nur auf predict verlassen, sondern einen expliziten Schwellenwert aus der Score-Verteilung der Trainingsdaten speichern. Diesen können Sie später anpassen, ohne neu zu trainieren.

Schritt 4: Live-MQTT-Nachrichten bewerten

Der folgende Dienst abonniert die Telemetrie der Pumpe, hält die letzten zehn Messwerte vor, berechnet dieselben Merkmale und veröffentlicht bei jedem Messwert einen Score.

# live_score.py
import json
import time
import uuid
from collections import deque

import joblib
import pandas as pd
import paho.mqtt.client as mqtt

from features import SIGNALS, WINDOW, window_features

PREFIX = "connectmq/demo/if-5d21/site1/pumps/pump1"
bundle = joblib.load("iforest_pump.joblib")
model, columns, threshold = bundle["model"], bundle["columns"], bundle["threshold"]
buffer = deque(maxlen=WINDOW)
streak = 0


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):
    global streak
    try:
        reading = json.loads(msg.payload)
        buffer.append({s: float(reading[s]) for s in SIGNALS})
    except (ValueError, KeyError):
        return
    if len(buffer) < WINDOW:
        return
    feats = window_features(pd.DataFrame(buffer))
    X = pd.DataFrame([feats], columns=columns)
    score = float(-model.score_samples(X)[0])
    is_anomaly = score > threshold
    streak = streak + 1 if is_anomaly else 0
    result = {"ts": time.time(), "score": round(score, 4),
              "threshold": round(threshold, 4), "isAnomaly": is_anomaly,
              "model": bundle["version"],
              "flowPerAmp": round(feats["flow_per_amp"], 3)}
    client.publish(f"{PREFIX}/anomaly", json.dumps(result), qos=0)
    if streak == 3:
        client.publish(f"{PREFIX}/alerts", json.dumps(result), qos=1)
        print("ALERT", result)


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

Schritt 5: Normale und fehlerhafte Daten einspeisen

Dieser Publisher spielt normales Verhalten ab und simuliert nach einer Minute ein blockiertes Laufrad: Der Strom bleibt hoch, während der Durchfluss sinkt.

# replay.py
import json
import time
import uuid

import numpy as np
import paho.mqtt.client as mqtt

PREFIX = "connectmq/demo/if-5d21/site1/pumps/pump1"
rng = np.random.default_rng()
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                     client_id=f"pump-sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()

for t in range(240):
    load = 0.6 + rng.normal(0, 0.02)
    reading = {
        "currentA": 10 + 12 * load + rng.normal(0, 0.3),
        "flowLpm": 30 + 40 * load + rng.normal(0, 1.0),
        "pressureBar": 2 + 3 * load + rng.normal(0, 0.08),
    }
    if t > 60:
        reading["flowLpm"] *= 0.75  # blockiertes Laufrad
    client.publish(f"{PREFIX}/telemetry",
                   json.dumps({k: round(v, 3) for k, v in reading.items()}))
    time.sleep(1)
client.loop_stop()

Starten Sie live_score.py und danach replay.py. In der ersten Minute bleiben die Scores unter dem Schwellenwert und steigen steil an, sobald der Durchfluss sinkt, weil die Beziehung zwischen Durchfluss und Strom zusammenbricht, während Strom und Druck noch unauffällig aussehen.

Einen Alarm für den Techniker erklären

Ein Anomalie-Score allein sagt niemandem, was zu prüfen ist. Isolation Forest bietet keine eingebaute Erklärung pro Merkmal, doch eine günstige Näherung funktioniert in der Praxis gut: Berechnen Sie für jedes Merkmal, wie weit der aktuelle Wert von seiner Trainingsverteilung entfernt liegt, zum Beispiel als robusten Z-Score gegenüber Median und Interquartilsabstand der Trainingsdaten, und nehmen Sie die zwei oder drei extremsten Merkmale in den Alarm auf.

Ein Alarm wie „Score 0,71 über Schwellenwert 0,62; flow_per_amp ungewöhnlich niedrig, flowLpm_mean ungewöhnlich niedrig“ weist direkt auf das Laufrad oder die Saugseite hin. Das ist weit hilfreicher als „Anomalie pump1“. Benötigen Sie belastbarere Zuordnungen, lassen sich für Baumensembles SHAP-Werte berechnen; sie kosten jedoch mehr Rechenleistung und sind für eine erste Bereitstellung selten nötig.

Hilfreich ist es auch, die Merkmalswerte selbst mit geringer Rate auf dem Anomalie-Topic zu veröffentlichen. Untersucht ein Ingenieur den Fall später, sieht er die Eingaben, die das Modell gesehen hat, und nicht nur dessen Urteil.

Checkliste zur Abstimmung von Isolation Forest auf Sensordaten

1. Auf repräsentativen Normaldaten trainieren. Berücksichtigen Sie alle regulären Betriebsarten, Schichten und Jahreszeiten, die Sie erwarten.
2. Bekannte Störungen vom Training ausschließen oder, falls das nicht möglich ist, die Kontamination ehrlich angeben.
3. Merkmale physikalisch sinnvoll halten. Verhältnisse und Differenzen schlagen oft Rohwerte.
4. Zeitstempel und Zähler nicht als Merkmale verwenden; sie lassen jeden neuen Messwert neuartig erscheinen.
5. Den Schwellenwert anhand von Validierungsdaten festlegen und nach den ersten Wochen mit Alarmen überprüfen.
6. Persistenz verlangen, bevor ein Alarm ausgelöst wird, wie es der Zähler streak oben tut.
7. Das Modell versionieren und die Version in jedes veröffentlichte Ergebnis aufnehmen.
8. Planmäßig neu trainieren oder dann, wenn das Drift-Monitoring eine Verschiebung der Eingangsverteilung meldet.

Am letzten Punkt trifft die Anomalieerkennung auf MLOps für IoT: Der Detektor ist selbst ein Modell, das überwacht werden muss.

Betrieb in der Produktion

Verlagern Sie den Dienst auf ein Edge-Gateway oder einen Server nahe an Ihren Daten, verbinden Sie sich über TLS auf Port 8883 mit den von Ihrem Broker ausgegebenen Zugangsdaten und behalten Sie die Topic-Struktur bei. In einem Werk mit Unified Namespace (UNS) liegt das Anomalie-Topic direkt neben der Telemetrie im ISA-95-Pfad der Anlage, etwa acme/werk-muenchen/versorgung/pumpen/pumpe-1/anomaly, sodass MES, Instandhaltung und Predictive-Maintenance-Dashboards den Score ohne zusätzliche Zuordnung finden. Ein privater ConnectMQ-Broker bietet Ihnen einen Live-Topic-Explorer, in dem Sie Scores neben der Rohtelemetrie verfolgen, und Schwellenwert-Alarmregeln im Portal können einen numerischen score über einer Grenze als einfaches zweites Sicherheitsnetz markieren. Wenn Sie überhaupt keine Broker betreiben möchten, übernimmt ein verwalteter MQTT-Broker wie ConnectMQ Zertifikate, Authentifizierung und Verfügbarkeit, während Ihr Modell Ihr eigener Code bleibt. Für Prototypen ohne Konto genügt der öffentliche MQTT-Broker broker.connectmq.com.

Die umgebende Architektur beschreibt Methoden der IoT-Anomalieerkennung; eine schlankere Einzelsignal-Version dieses Dienstes finden Sie unter Echtzeit-Anomalieerkennung in Python. Den größeren Zusammenhang erläutert der Leitfaden zu KI und IoT.

Mit ConnectMQ starten

Erstellen Sie unter app.connectmq.com einen Broker im kostenlosen Serverless-Tarif, ganz ohne Kreditkarte, und ersetzen Sie in live_score.py den öffentlichen Host durch Ihren eigenen Endpunkt und Ihre Zugangsdaten. Ihre Isolation-Forest-Ergebnisse erscheinen im Topic-Explorer, sobald das erste Fenster gefüllt ist.

Häufig gestellte Fragen

Wie viele Daten brauche ich, um Isolation Forest auf Sensordaten zu trainieren?

Für den Einstieg reichen meist einige tausend Merkmalsfenster, die alle normalen Betriebsarten abdecken. Die Abdeckung der Betriebsarten ist wichtiger als die reine Datenmenge.

Welchen Wert für die Kontamination sollte ich verwenden?

Verwenden Sie Ihre beste Schätzung, wie viel der Trainingsdaten anomal ist – bei kuratierten Normaldaten oft deutlich unter einem Prozent. Da der Live-Dienst einen eigenen Schwellenwert nutzt, wirkt sich die Kontamination hauptsächlich auf die Standardausgabe von predict aus.

Kann Isolation Forest auf einem Mikrocontroller laufen?

Die trainierten Bäume sind klein, doch die Merkmalsberechnung und scikit-learn selbst erfordern ein Gerät der Linux-Klasse. Betreiben Sie das Modell auf einem Gateway oder Raspberry Pi und belassen Sie Mikrocontroller bei einfachen Schwellenwerten.

Warum markiert mein Modell nach einer Prozessänderung alles?

Isolation Forest bewertet relativ zu seinen Trainingsdaten, daher wirkt ein neues Produkt oder ein neuer Sollwert anomal. Trainieren Sie mit Daten aus der neuen Betriebsart neu oder trainieren Sie separate Modelle pro Betriebsart.