IoT-Zeitreihenprognose mit Python und MQTT

Die IoT-Zeitreihenprognose sagt die künftigen Werte eines Sensorsignals wie Temperatur, Tankfüllstand, Durchfluss oder Leistung aus seiner eigenen Historie und verwandten Eingangsgrößen voraus. In diesem Tutorial erfassen Sie mit Python Sensordaten von einem MQTT-Broker, tasten sie zu einer sauberen Reihe um, vergleichen in einem korrekten Backtest ein statistisches Modell mit einem Machine-Learning-Modell und publizieren die Prognose zurück an MQTT, damit Dashboards und Steuerungen sie nutzen können.

Warum Sensordaten prognostizieren

Prognosen machen aus Überwachung Planung. Eine Prognose der Kühlraumtemperatur sagt Ihnen, ob der Verdichter die Nachmittagsspitze schafft. Eine Prognose des Tankfüllstands sagt Ihnen, wann Sie die nächste Lieferung bestellen müssen. Eine Prognose des Energiebedarfs sagt Ihnen, wann Sie Lasten verschieben sollten, etwa um in einem Werk teure Lastspitzen zu vermeiden. In jedem Fall liegt der Nutzen darin, zu handeln, bevor der Messwert eine Grenze erreicht.

Prognose ist etwas anderes als Anomalieerkennung. Die Anomalieerkennung fragt, ob der aktuelle Wert ungewöhnlich ist; die Prognose fragt, wie der Wert später sein wird. Beide arbeiten oft zusammen: Eine große Lücke zwischen Prognose und Messung ist selbst ein nützliches Anomaliesignal, wie in IoT-Anomalieerkennung beschrieben.

Was die IoT-Zeitreihenprognose knifflig macht

Sensordaten sind unordentlicher als die aufgeräumten Datensätze der meisten Prognose-Tutorials:

Der größte Teil der Arbeit in einem Prognoseprojekt besteht in der Vorbereitung der Reihe; das Modell ist der kleinere Teil. Der Artikel zu industriellen Zeitreihendaten behandelt Praktiken für Abtastung und Zeitstempel, die diese Vorbereitung deutlich erleichtern.

Horizont, Auflösung und Eingangsgrößen wählen

Bevor Sie Code schreiben, legen Sie drei Entscheidungen fest, denn sie prägen alles Weitere.

Horizont. Wie weit muss die Entscheidung vorausblicken? Eine Kühlungsregelung braucht vielleicht 30 Minuten, ein Lieferplaner mehrere Tage. Der Fehler wächst mit dem Horizont, prognostizieren Sie also nicht weiter, als die Entscheidung es erfordert. Wenn Sie sowohl kurze als auch lange Horizonte brauchen, ist es oft besser, getrennte Modelle zu trainieren, als eines zu überdehnen.

Auflösung. Der Prognoseschritt sollte dazu passen, wie oft sich die Entscheidung ändern kann. Einen Tankfüllstand sekündlich zu prognostizieren ist sinnlos, wenn der Liefer-Lkw einmal am Tag gebucht wird. Eine gröbere Auflösung glättet zudem das Sensorrauschen und reduziert die Anzahl der Schritte, die das Modell vorhersagen muss. Für die meisten Gebäude- und Prozesssignale sind 5 bis 60 Minuten ein sinnvoller Bereich.

Eingangsgrößen. Entscheiden Sie, welche externen Treiber Sie im Voraus kennen können. Kalender-Features sind immer verfügbar. Wetterprognosen, Produktionspläne und Belegungspläne oft auch. Messwerte anderer Sensoren sind nur bis zur Gegenwart verfügbar und können daher nur als verzögerte Eingangsgrößen genutzt werden. Schreiben Sie diese Liste auf; sie ist der Vertrag, der später im Projekt Leakage verhindert.

Schritt 1: Sensordaten aus MQTT erfassen

Für ein in sich geschlossenes Tutorial simuliert dieses Skript einen Temperatursensor mit Tageszyklus und Rauschen, publiziert ihn an den öffentlichen MQTT-Broker, der für Tests und Prototypen gedacht ist, und zeichnet ihn mit einem Subscriber in einer CSV-Datei auf. Simulierte „Minuten“ werden gestaucht, damit der Lauf schnell endet; die Zeitstempel in der Payload tragen die simulierte Zeit.

# capture.py
import csv
import json
import math
import random
import threading
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
TOPIC = f"connectmq/demo/{uuid.uuid4().hex[:8]}/coldroom1/temperature"
START = 1767225600  # 2026-01-01T00:00:00Z
STEPS = 14 * 24 * 4  # 14 Tage in 15-Minuten-Auflösung
received = []


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    p = json.loads(msg.payload)
    received.append((p["ts"], p["value"]))


def simulate(pub):
    for i in range(STEPS):
        t = START + i * 900
        hour = (t % 86400) / 3600
        weekend = time.gmtime(t).tm_wday >= 5
        value = 4.0 + 1.2 * math.sin((hour - 9) / 24 * 2 * math.pi) + (-0.4 if weekend else 0)
        value += random.gauss(0, 0.15)
        if random.random() < 0.02:
            continue  # einen verlorenen Messwert simulieren
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # Limit des öffentlichen Brokers von 10 msg/s pro Client einhalten


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-cap-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()
time.sleep(1)
simulate(client)
time.sleep(3)
client.loop_stop()

with open("coldroom.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["ts", "value"])
    w.writerows(sorted(received))
print("rows:", len(received), "topic:", TOPIC)

In der Produktion kämen dieselben Daten aus Ihrem Historian oder von einem Subscriber, der mit Ihrem eigenen ConnectMQ-Broker über YOUR_USERNAME, YOUR_PASSWORD und YOUR_CLIENT_ID per TLS auf Port 8883 verbunden ist.

Schritt 2: Die Reihe vorbereiten

Laden Sie die CSV-Datei, konvertieren Sie die Zeitstempel, tasten Sie auf ein festes 15-Minuten-Raster um und interpolieren Sie nur kurze Lücken. Längere Lücken sollten fehlen bleiben, damit sie keine falsche Historie erzeugen.

# prepare.py
import pandas as pd

df = pd.read_csv("coldroom.csv")
df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
s = df.set_index("ts")["value"].resample("15min").mean()
s = s.interpolate(limit=4)  # Lücken bis zu einer Stunde füllen
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Schritt 3: Zwei Modelle für die IoT-Zeitreihenprognose

Eine gute Gewohnheit ist es, immer mindestens drei Prognosen zu vergleichen: eine naive Baseline, ein statistisches Modell und ein Machine-Learning-Modell. Wenn ein komplexes Modell „gleiche Uhrzeit gestern“ nicht schlägt, lohnt sich seine Bereitstellung nicht.

# forecast.py
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.metrics import mean_absolute_error
from statsmodels.tsa.holtwinters import ExponentialSmoothing

s = pd.read_parquet("coldroom.parquet")["value"].ffill()
DAY = 96          # 15-Minuten-Schritte pro Tag
HORIZON = DAY     # 24 Stunden im Voraus prognostizieren


def make_features(series):
    X = pd.DataFrame(index=series.index)
    X["lag_day"] = series.shift(DAY)
    X["lag_day2"] = series.shift(2 * DAY)
    X["roll_day_mean"] = series.shift(DAY).rolling(DAY).mean()
    X["hour"] = series.index.hour + series.index.minute / 60
    X["dow"] = series.index.dayofweek
    return X


results = {"naive": [], "holt_winters": [], "gbr": []}
# Rolling-Origin-Backtest: jeden der letzten 4 Tage aus den Daten davor prognostizieren
for k in range(4, 0, -1):
    cut = len(s) - k * DAY
    train, test = s.iloc[:cut], s.iloc[cut:cut + HORIZON]

    naive = train.iloc[-DAY:].values
    results["naive"].append(mean_absolute_error(test, naive))

    hw = ExponentialSmoothing(train, trend=None, seasonal="add", seasonal_periods=DAY).fit()
    results["holt_winters"].append(mean_absolute_error(test, hw.forecast(HORIZON)))

    X = make_features(s)
    Xtr, ytr = X.iloc[:cut].dropna(), s.iloc[:cut][X.iloc[:cut].dropna().index]
    gbr = HistGradientBoostingRegressor(max_iter=300, learning_rate=0.05).fit(Xtr, ytr)
    results["gbr"].append(mean_absolute_error(test, gbr.predict(X.iloc[cut:cut + HORIZON])))

for name, maes in results.items():
    print(f"{name:13s} MAE {np.mean(maes):.3f}")

Installieren Sie die Bibliotheken mit pip install paho-mqtt pandas pyarrow scikit-learn statsmodels. Das Gradient-Boosting-Modell verwendet nur Lags von mindestens einem Tag, sodass jedes Feature zum Prognosezeitpunkt für den gesamten 24-Stunden-Horizont bekannt ist. Das ist der häufigste Fehler bei ML-Prognosen: im Training einen Lag von einem Schritt zu verwenden und dann bei weiter vorausreichenden Prognosen nichts zu haben, womit man ihn füttern kann.

Schritt 4: Ehrlich backtesten

Die obige Schleife ist ein Rolling-Origin-Backtest: Jeder Fold trainiert nur auf Daten vor seinem Stichtag und prognostiziert den folgenden Tag. Mischen Sie Zeitreihen vor dem Aufteilen niemals. Geben Sie den Fehler in der Einheit des Signals an (ein MAE in Grad lässt sich leichter diskutieren als ein Prozentwert nahe null), und betrachten Sie die Fehler nach Tageszeit, denn ein Modell kann im Mittel gut und genau in der Spitze, auf die es ankommt, schlecht sein.

Checkliste für eine vertrauenswürdige Evaluation:

1. Verwenden Sie mehrere Prognoseursprünge, nicht einen einzigen Split.
2. Vergleichen Sie mit einer naiven saisonalen Baseline.
3. Stellen Sie sicher, dass jedes Feature zum Prognosezeitpunkt verfügbar ist.
4. Prüfen Sie die Fehler rund um Ereignisse wie Abtauzyklen, Feiertage oder Stillstände.
5. Führen Sie den Backtest nach jeder Änderung am Vorbereitungscode erneut aus.

Schritt 5: Prognosen zurück an MQTT publizieren

Eine Prognose ist dort am nützlichsten, wo Entscheidungen fallen, also publizieren Sie sie neben dem Messwert. Eine Retained Message sorgt dafür, dass jeder neue Subscriber sofort die aktuelle Prognose erhält. In einem Werk mit Unified Namespace (UNS) bedeutet „neben dem Messwert“ konkret: im selben ISA-95-Pfad, etwa acme/werk-stuttgart/montage/linie-1/zelle-3/temperatur und .../zelle-3/temperatur-prognose, sodass MES, Energiemonitoring und Dashboards Ist und Prognose ohne zusätzliches Mapping finden.

# publish_forecast.py
import json
import time
import uuid

import pandas as pd
import paho.mqtt.client as mqtt
from statsmodels.tsa.holtwinters import ExponentialSmoothing

s = pd.read_parquet("coldroom.parquet")["value"].ffill()
model = ExponentialSmoothing(s, seasonal="add", seasonal_periods=96).fit()
fc = model.forecast(96)

payload = {
    "generated": int(time.time() * 1000),
    "model": "holt-winters-v1",
    "step_minutes": 15,
    "points": [{"ts": int(t.timestamp() * 1000), "value": round(v, 2)} for t, v in fc.items()],
}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-pub-" + uuid.uuid4().hex[:8])
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
info = client.publish("connectmq/demo/forecast-demo/coldroom1/forecast", json.dumps(payload), qos=1, retain=True)
info.wait_for_publish()
client.loop_stop()
client.disconnect()
print(len(json.dumps(payload)), "bytes published")

Eine Prognose mit 96 Punkten umfasst einige Kilobyte und liegt damit deutlich unter dem Nachrichtenlimit des öffentlichen Brokers von 64 KB und dem Limit von 256 KB im kostenlosen Serverless-Plan. Ändern Sie das Demo-Topic-Segment vor dem Ausführen in etwas Eindeutiges. Nachgelagert stellt ein Dashboard forecast gegen temperature dar, und ein Steuerungsskript kann eingreifen, wenn die Prognose in den nächsten Stunden eine Grenze überschreitet.

Dauerhaft betreiben

Machen Sie aus den Skripten einen kleinen Dienst: täglich auf den letzten Wochen neu trainieren, alle 15 Minuten prognostizieren und publizieren. Protokollieren Sie den Prognosefehler, sobald die Istwerte eintreffen, und publizieren Sie ihn als eigenes Topic, damit Sie bemerken, wenn die Genauigkeit nachlässt; das ist der Anfang der Monitoring-Praktiken, die in IoT-MLOps beschrieben sind. Eine domänenspezifische Version derselben Pipeline finden Sie unter Energielastprognose.

Starten Sie mit ConnectMQ

Erstellen Sie einen kostenlosen Serverless-Broker unter app.connectmq.com, ohne Kreditkarte, und verbinden Sie Ihre Sensoren und Ihren Prognosedienst mit den generierten Zugangsdaten. ConnectMQ betreibt den verwalteten MQTT-Broker für Sie, und mit dem Live-Topic-Explorer prüfen Sie Messwerte und Prognosen nebeneinander. Für das Gesamtbild, wie Modelle und Geräte zusammenkommen, lesen Sie den Leitfaden zu KI und IoT, und wenn Sie skalieren, unterstützt ein Cloud-MQTT-Broker-Plan mehr Verbindungen und größere Nachrichten.

Häufig gestellte Fragen

Welches Modell ist für die IoT-Zeitreihenprognose am besten?

Es gibt nicht das eine beste Modell. Beginnen Sie mit einer naiven saisonalen Baseline, vergleichen Sie dann ein Exponential-Smoothing- oder ARIMA-Modell mit einem Gradient-Boosting-Modell auf Lag- und Kalender-Features, und behalten Sie dasjenige, das im Rolling-Backtest gewinnt.

Wie viel Historie brauche ich?

Mindestens mehrere vollständige Zyklen der längsten Saisonalität, die Sie erfassen wollen, zum Beispiel einige Wochen für Tagesmuster und mehrere Monate für Wochenmuster mit Feiertagen.

Wie gehe ich mit fehlenden Sensorwerten um?

Tasten Sie auf ein festes Intervall um, interpolieren Sie nur kurze Lücken und lassen Sie lange Ausfälle fehlen. Modelle, die auf langen Strecken erfundener Daten trainiert werden, lernen Muster, die es nie gab.

Wo sollte das Prognosemodell laufen?

Meist auf einem Server oder in der Cloud, als MQTT-Client, der Messwerte abonniert und Prognosen publiziert. Kleine Modelle können auch auf Edge-Gateways nahe an der Anlage laufen.