Isolation Forest pour les données de capteurs : entraîner hors ligne, évaluer les messages MQTT en direct

Appliquer Isolation Forest aux données de capteurs est l'un des moyens les plus pratiques de détecter des anomalies cachées dans la combinaison de plusieurs signaux. Dans ce tutoriel, vous allez générer des données historiques, construire des variables par fenêtre, entraîner un modèle avec scikit-learn et l'exécuter dans un service Python qui évalue les messages MQTT en direct et publie des événements d'anomalie.

Pourquoi Isolation Forest fonctionne bien sur les données de capteurs

Isolation Forest repose sur une idée simple : les anomalies sont rares et différentes, donc plus faciles à isoler. L'algorithme construit de nombreux arbres aléatoires ; chaque arbre choisit à répétition une variable et une valeur de coupure au hasard. Les points normaux se situent dans des régions denses et nécessitent de nombreuses coupures pour être séparés. Les points anormaux sont isolés après seulement quelques coupures. La longueur moyenne du chemin à travers les arbres devient le score d'anomalie.

Cette conception convient aux capteurs industriels pour plusieurs raisons :

• Aucun label nécessaire. Vous entraînez le modèle sur un fonctionnement majoritairement normal, ce dont chaque usine dispose en abondance.
• Multivarié par nature. Un courant élevé est normal, un faible débit est normal, mais un courant élevé avec un faible débit est isolé rapidement.
• Rapide. L'entraînement sur des centaines de milliers de lignes prend quelques secondes, et l'évaluation est assez légère pour une passerelle edge.
• Peu d'hyperparamètres. Le nombre d'arbres, la taille d'échantillon et la contamination attendue sont généralement les seuls réglages.

Il a aussi ses limites. Il ne comprend pas le temps par lui-même : vous devez donc encoder le comportement temporel dans les variables. Et il évalue par rapport aux données d'entraînement : un modèle entraîné en été peut signaler l'hiver comme anormal.

Isolation Forest face aux autres détecteurs

Si vous n'avez pas encore essayé les statistiques par signal, lisez d'abord la détection de valeurs aberrantes dans les données de capteurs avec Z-score, IQR et EWMA. Isolation Forest prend tout son intérêt lorsque ces méthodes passent à côté des relations entre signaux.

Prérequis

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

Choisissez un préfixe aléatoire comme connectmq/demo/if-5d21/.... Les exemples utilisent le serveur MQTT public gratuit broker.connectmq.com, réservé aux tests et aux prototypes, où tous les topics sont lisibles par d'autres : n'utilisez donc que des données synthétiques.

Étape 1 : créer des données de capteurs historiques

Dans un vrai projet, vous exporteriez l'historique depuis votre base de données. Pour que le tutoriel soit autonome, ce script simule une pompe avec trois signaux corrélés : courant moteur, débit et pression de refoulement.

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

rng = np.random.default_rng(42)
n = 20_000  # une mesure par seconde, environ 5,5 heures
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())

Les trois signaux évoluent ensemble avec la charge. C'est précisément cette corrélation qu'un seuil par signal ne peut pas voir.

Étape 2 : construire des variables par fenêtre

Évaluer des mesures isolées ignore le bruit et les dynamiques courtes. Calculez plutôt des variables sur une courte fenêtre glissante : la moyenne de chaque signal, son écart-type et quelques ratios ayant un sens physique.

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

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


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)

Les ratios comptent plus qu'il n'y paraît. Une roue obstruée fait chuter le débit par ampère, et une fuite fait chuter la pression par unité de débit, même lorsque chaque valeur brute reste dans sa plage normale.

Étape 3 : entraîner l'Isolation Forest

# 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,   # part attendue d'anomalies dans les données d'entraînement
    random_state=0,
)
model.fit(X)

scores = -model.score_samples(X)        # plus élevé = plus anormal
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}")

Deux détails méritent attention. D'abord, la méthode score_samples de scikit-learn renvoie des valeurs plus faibles pour les points les plus anormaux : en prendre l'opposé donne un score intuitif où « plus élevé signifie pire ». Ensuite, au lieu de vous fier uniquement à predict, stockez un seuil explicite tiré de la distribution des scores d'entraînement. Vous pourrez ainsi l'ajuster plus tard sans réentraîner.

Étape 4 : évaluer les messages MQTT en direct

Le service ci-dessous s'abonne à la télémétrie de la pompe, conserve les dix dernières mesures, calcule les mêmes variables et publie un score à chaque mesure.

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

Étape 5 : l'alimenter en données normales et défaillantes

Ce publieur rejoue un comportement normal puis, au bout d'une minute, simule une roue obstruée : le courant reste élevé tandis que le débit chute.

# 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  # roue obstruée
    client.publish(f"{PREFIX}/telemetry",
                   json.dumps({k: round(v, 3) for k, v in reading.items()}))
    time.sleep(1)
client.loop_stop()

Lancez live_score.py, puis replay.py. Les scores restent sous le seuil pendant la première minute et grimpent nettement dès que le débit chute, car la relation débit par ampère se rompt alors que le courant et la pression semblent encore ordinaires.

Expliquer une alerte à un technicien

Un score d'anomalie seul n'indique à personne ce qu'il faut vérifier. Isolation Forest n'offre pas d'explication native par variable, mais une approximation peu coûteuse fonctionne bien en pratique : pour chaque variable, calculez à quelle distance la valeur actuelle se situe de sa distribution d'entraînement, par exemple sous forme de z-score robuste par rapport à la médiane et à l'écart interquartile d'entraînement, et incluez dans l'alerte les deux ou trois variables les plus extrêmes.

Une alerte qui indique « score 0,71 au-dessus du seuil 0,62 ; flow_per_amp anormalement bas, flowLpm_mean anormalement bas » oriente directement vers la roue ou le côté aspiration. C'est bien plus exploitable que « anomalie pump1 ». Si vous avez besoin d'attributions plus rigoureuses, les valeurs SHAP peuvent être calculées pour les ensembles d'arbres, mais elles demandent plus de calcul et sont rarement nécessaires pour un premier déploiement.

Il est également utile de publier les valeurs des variables elles-mêmes sur le topic d'anomalies, à faible cadence. Lorsqu'un ingénieur enquête plus tard, il peut voir les entrées qu'a vues le modèle, et pas seulement son verdict.

Liste de contrôle du réglage d'Isolation Forest sur les données de capteurs

1. Entraînez sur des données normales représentatives. Incluez tous les modes de fonctionnement réguliers, les équipes et les saisons que vous vous attendez à rencontrer.
2. Excluez les défauts connus de l'entraînement, ou gardez une contamination réaliste si vous ne le pouvez pas.
3. Gardez des variables ayant un sens physique. Les ratios et les différences battent souvent les valeurs brutes.
4. Évitez les horodatages et les compteurs comme variables ; ils font paraître chaque nouvelle mesure inédite.
5. Fixez le seuil à partir de données de validation et revoyez-le après les premières semaines d'alertes.
6. Exigez une persistance avant d'alerter, comme le fait le compteur streak ci-dessus.
7. Versionnez le modèle et incluez la version dans chaque résultat publié.
8. Réentraînez selon un calendrier ou lorsque la surveillance de dérive indique que la distribution des entrées a changé.

Ce dernier point est celui où la détection d'anomalies rejoint le MLOps pour l'IoT : le détecteur est lui-même un modèle qui doit être surveillé.

Passer en production

Déplacez le service sur une passerelle edge ou un serveur proche de vos données, connectez-vous en TLS sur le port 8883 avec des identifiants émis par votre broker et conservez la structure des topics. Un broker ConnectMQ privé vous offre un explorateur de topics en direct pour suivre les scores à côté de la télémétrie brute, et des règles d'alerte à seuil dans le portail peuvent signaler un score numérique au-dessus d'une limite, comme second filet de sécurité simple. Si vous préférez ne pas exploiter de broker du tout, un broker MQTT managé comme ConnectMQ gère les certificats, l'authentification et la disponibilité, tandis que votre modèle reste votre propre code. Pour prototyper sans compte, le serveur MQTT public gratuit broker.connectmq.com suffit.

Pour l'architecture d'ensemble, consultez les méthodes de détection d'anomalies IoT ; pour une version plus légère de ce service, limitée à un signal, consultez la détection d'anomalies en temps réel en Python. Le contexte plus large se trouve dans le guide IA et IoT.

Démarrer avec ConnectMQ

Créez un broker avec l'offre gratuite Serverless sur app.connectmq.com, sans carte bancaire, et remplacez l'hôte public dans live_score.py par votre propre endpoint et vos identifiants. Les résultats de votre Isolation Forest apparaîtront dans l'explorateur de topics dès que la première fenêtre sera remplie.

Questions fréquentes

De combien de données ai-je besoin pour entraîner Isolation Forest sur des données de capteurs ?

Quelques milliers de fenêtres de variables couvrant tous les modes de fonctionnement normaux suffisent généralement pour commencer. La couverture des modes compte davantage que le volume brut.

Quelle valeur de contamination utiliser ?

Utilisez votre meilleure estimation de la part anormale des données d'entraînement, souvent bien inférieure à un pour cent pour des données normales soigneusement sélectionnées. Comme le service en direct utilise son propre seuil, la contamination affecte surtout la sortie par défaut de predict.

Isolation Forest peut-il s'exécuter sur un microcontrôleur ?

Les arbres entraînés sont petits, mais le calcul des variables et scikit-learn lui-même nécessitent un équipement de classe Linux. Exécutez-le sur une passerelle ou un Raspberry Pi et réservez aux microcontrôleurs des seuils simples.

Pourquoi mon modèle signale-t-il tout après une modification du procédé ?

Isolation Forest évalue par rapport à ses données d'entraînement : un nouveau produit ou une nouvelle consigne paraît donc anormal. Réentraînez-le avec des données du nouveau mode, ou entraînez des modèles distincts par mode.