Prévision de séries temporelles IoT avec Python et MQTT

La prévision de séries temporelles IoT prédit les valeurs futures d'un signal de capteur, comme une température, un niveau de cuve, un débit ou une puissance, à partir de son propre historique et d'entrées associées. Dans ce tutoriel, vous allez capter des données de capteurs depuis un broker MQTT avec Python, les rééchantillonner en une série propre, comparer un modèle statistique à un modèle de machine learning à l'aide d'un véritable backtest, puis republier la prévision sur MQTT pour que tableaux de bord et contrôleurs puissent l'utiliser.

Pourquoi prévoir les données de capteurs

Les prévisions transforment la supervision en planification. Une prévision de la température d'une chambre froide indique si le compresseur suivra pendant le pic de l'après-midi. Une prévision du niveau d'une cuve indique quand commander la prochaine livraison. Une prévision de la demande énergétique indique quand décaler des charges. Dans chaque cas, la valeur réside dans le fait d'agir avant que la mesure n'atteigne une limite.

La prévision diffère de la détection d'anomalies. La détection d'anomalies demande si la valeur actuelle est inhabituelle ; la prévision demande quelle sera la valeur plus tard. Les deux fonctionnent souvent ensemble : un écart important entre prévision et mesure est en soi un signal d'anomalie utile, comme expliqué dans l'article sur la détection d'anomalies IoT.

Ce qui complique la prévision de séries temporelles IoT

Les données de capteurs sont plus désordonnées que les jeux de données bien rangés de la plupart des tutoriels de prévision :

L'essentiel du travail d'un projet de prévision consiste à préparer la série ; le modèle en est la plus petite partie. L'article sur les séries temporelles industrielles couvre les pratiques d'échantillonnage et d'horodatage qui facilitent grandement cette préparation.

Choisir l'horizon, la résolution et les entrées

Avant d'écrire du code, arrêtez trois décisions, car elles conditionnent tout le reste.

L'horizon. Jusqu'où la décision doit-elle voir ? Un régulateur de froid peut avoir besoin de 30 minutes, un planificateur de livraisons de plusieurs jours. L'erreur croît avec l'horizon : ne prévoyez donc pas plus loin que ce que la décision exige. Si vous avez besoin d'horizons courts et longs, il vaut souvent mieux entraîner des modèles distincts que d'en étirer un seul.

La résolution. Le pas de prévision doit correspondre à la fréquence à laquelle la décision peut changer. Prévoir un niveau de cuve à la seconde n'a aucun sens si le camion de livraison est réservé une fois par jour. Une résolution plus grossière lisse aussi le bruit des capteurs et réduit le nombre de pas que le modèle doit prédire. Pour la plupart des signaux de bâtiment et de process, une plage de 5 à 60 minutes est raisonnable.

Les entrées. Déterminez quels facteurs externes vous pouvez connaître à l'avance. Les features calendaires sont toujours disponibles. Les prévisions météo, les plans de production et les plannings d'occupation le sont souvent. Les valeurs mesurées d'autres capteurs ne sont disponibles que jusqu'à l'instant présent, elles ne peuvent donc servir que d'entrées décalées. Consignez cette liste : c'est le contrat qui évitera les fuites de données plus tard dans le projet.

Étape 1 : capter les données de capteurs depuis MQTT

Pour un tutoriel autonome, ce script simule un capteur de température avec un cycle journalier et du bruit, le publie sur le serveur MQTT public gratuit et l'enregistre en retour via un abonné dans un fichier CSV. Les « minutes » simulées sont compressées pour que l'exécution se termine rapidement ; les horodatages du payload portent le temps simulé.

# 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 jours à une résolution de 15 minutes
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  # simuler une mesure perdue
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # respecter la limite de 10 msg/s par client du broker public


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)

En production, ces mêmes données proviendraient de votre historian, ou d'un abonné connecté à votre propre broker ConnectMQ avec YOUR_USERNAME, YOUR_PASSWORD et YOUR_CLIENT_ID, en TLS sur le port 8883.

Étape 2 : préparer la série

Chargez le CSV, convertissez les horodatages, rééchantillonnez sur une grille fixe de 15 minutes et n'interpolez que les trous courts. Les trous plus longs doivent rester vides pour ne pas créer un faux historique.

# 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)  # combler les trous jusqu'à une heure
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Étape 3 : deux modèles de prévision de séries temporelles IoT

Une bonne habitude consiste à toujours comparer au moins trois prévisions : une référence naïve, un modèle statistique et un modèle de machine learning. Si un modèle complexe ne bat pas « même heure hier », il ne vaut pas la peine d'être déployé.

# 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          # pas de 15 minutes par jour
HORIZON = DAY     # prévoir 24 heures à l'avance


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": []}
# Backtest à origine glissante : prévoir chacun des 4 derniers jours à partir des données antérieures
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}")

Installez les bibliothèques avec pip install paho-mqtt pandas pyarrow scikit-learn statsmodels. Le modèle à gradient boosting n'utilise que des lags d'au moins un jour, si bien que chaque feature est connue au moment de la prévision sur tout l'horizon de 24 heures. C'est l'erreur la plus courante en prévision par ML : utiliser un lag d'un pas à l'entraînement, puis n'avoir rien pour l'alimenter lorsqu'on prévoit plus loin.

Étape 4 : réaliser un backtest honnête

La boucle ci-dessus est un backtest à origine glissante : chaque pli s'entraîne uniquement sur les données antérieures à sa date de coupure et prévoit le jour suivant. Ne mélangez jamais une série temporelle avant de la découper. Exprimez l'erreur dans l'unité du signal (une MAE en degrés se discute plus facilement qu'un pourcentage proche de zéro), et examinez les erreurs par heure de la journée, car un modèle peut être bon en moyenne et mauvais précisément au pic qui vous intéresse.

Checklist pour une évaluation fiable :

1. Utilisez plusieurs origines de prévision, pas un seul découpage.
2. Comparez à une référence saisonnière naïve.
3. Assurez-vous que chaque feature est disponible au moment de la prévision.
4. Examinez les erreurs autour des événements comme les cycles de dégivrage, les jours fériés ou les arrêts.
5. Relancez le backtest après toute modification du code de préparation.

Étape 5 : republier les prévisions sur MQTT

Une prévision est plus utile là où se prennent les décisions : publiez-la donc à côté de la mesure. Avec un retained message, tout nouvel abonné obtient immédiatement la dernière prévision.

# 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")

Une prévision de 96 points pèse quelques kilo-octets, bien en deçà de la limite de 64 Ko par message du serveur MQTT public gratuit et de la limite de 256 Ko de l'offre gratuite Serverless. Remplacez le segment de topic de démo par une valeur unique avant de l'exécuter. En aval, un tableau de bord trace forecast face à temperature, et un script de pilotage peut agir si la prévision franchit une limite dans les heures qui viennent.

Faire tourner le tout en continu

Transformez les scripts en un petit service : réentraînement quotidien sur les dernières semaines, prévision toutes les 15 minutes, puis publication. Consignez l'erreur de prévision dès que les valeurs réelles arrivent et publiez-la sur son propre topic pour remarquer toute dégradation de la précision ; c'est le point de départ des pratiques de supervision décrites dans l'article sur le MLOps IoT. Pour une version métier du même pipeline, consultez la prévision de la charge énergétique.

Démarrez avec ConnectMQ

Créez un broker Serverless gratuit sur app.connectmq.com, sans carte bancaire, et connectez vos capteurs et votre service de prévision avec les identifiants générés. ConnectMQ exploite pour vous le broker MQTT managé, et l'explorateur de topics en direct vous permet de vérifier mesures et prévisions côte à côte. Pour une vue d'ensemble de l'association entre modèles et appareils, lisez le guide IA et IoT, et lorsque vous passez à l'échelle, une offre de broker MQTT cloud prend en charge davantage de connexions et des messages plus volumineux.

Questions fréquentes

Quel est le meilleur modèle pour la prévision de séries temporelles IoT ?

Il n'existe pas de meilleur modèle unique. Commencez par une référence saisonnière naïve, puis comparez un modèle de lissage exponentiel ou ARIMA à un modèle à gradient boosting sur des features de lag et calendaires, et conservez celui qui l'emporte dans un backtest à origine glissante.

De combien d'historique ai-je besoin ?

D'au moins plusieurs cycles complets de la saisonnalité la plus longue que vous voulez capter, par exemple quelques semaines pour des motifs journaliers et plusieurs mois pour des motifs hebdomadaires avec jours fériés.

Comment gérer les mesures de capteurs manquantes ?

Rééchantillonnez à intervalle fixe, n'interpolez que les trous courts et laissez vides les longues coupures. Les modèles entraînés sur de longues périodes de données inventées apprennent des schémas qui n'ont jamais existé.

Où le modèle de prévision doit-il s'exécuter ?

Généralement sur un serveur ou dans le cloud, sous forme de client MQTT qui s'abonne aux mesures et publie les prévisions. Les petits modèles peuvent aussi tourner sur des passerelles edge au plus près des équipements.