Isolation Forest for Sensor Data: Train Offline, Score Live MQTT Messages

Isolation Forest on sensor data is one of the most practical ways to catch anomalies that hide in the combination of several signals. In this tutorial you will generate historical data, engineer window features, train a model with scikit-learn, and run it in a Python service that scores live MQTT messages and publishes anomaly events.

Why Isolation Forest Works Well for Sensor Data

Isolation Forest is built on a simple idea: anomalies are few and different, so they are easier to isolate. The algorithm builds many random trees; each tree repeatedly picks a random feature and a random split value. Normal points sit in dense regions and need many splits to be separated. Anomalous points are isolated after only a few. The average path length across trees becomes the anomaly score.

That design suits industrial sensors for several reasons:

• No labels needed. You train on mostly normal operation, which every plant has plenty of.
• Multivariate by nature. High current is normal, low flow is normal, but high current with low flow is isolated quickly.
• Fast. Training on hundreds of thousands of rows takes seconds and scoring is cheap enough for an edge gateway.
• Few hyperparameters. The number of trees, the sample size and the expected contamination are usually all you tune.

It has limits too. It does not understand time on its own, so you must encode temporal behavior in features. And it scores relative to the training data, so a model trained in summer may flag winter as anomalous.

Isolation Forest vs Other Detectors

If you have not yet tried single-signal statistics, read sensor data outlier detection with Z-score, IQR and EWMA first. Isolation Forest earns its place when those methods miss relationships between signals.

Prerequisites

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

Choose a random prefix such as connectmq/demo/if-5d21/.... The examples use the public test area of broker.connectmq.com, where all topics are readable by others, so keep the data synthetic.

Step 1: Create Historical Sensor Data

In a real project you would export history from your database. To keep the tutorial self-contained, this script simulates a pump with three correlated signals: motor current, flow and outlet pressure.

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

rng = np.random.default_rng(42)
n = 20_000  # one reading per second, about 5.5 hours
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())

All three signals move together with load. That correlation is exactly what a per-signal threshold cannot see.

Step 2: Engineer Window Features

Scoring single readings ignores noise and short dynamics. Instead, compute features over a short sliding window: the mean of each signal, its standard deviation, and a couple of physically meaningful ratios.

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

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


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)

The ratios matter more than they look. A blocked impeller drops flow per amp, and a leak drops pressure per flow, even when every raw value is within its normal range.

Step 3: Train the 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,   # expected share of anomalies in training data
    random_state=0,
)
model.fit(X)

scores = -model.score_samples(X)        # higher = more anomalous
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}")

Two details deserve attention. First, scikit-learn's score_samples returns lower values for more abnormal points, so negating it gives an intuitive "higher is worse" score. Second, instead of relying only on predict, store an explicit threshold taken from the training score distribution. You can then adjust it later without retraining.

Step 4: Score Live MQTT Messages

The service below subscribes to the pump's telemetry, keeps the last ten readings, computes the same features and publishes a score on every reading.

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

Step 5: Feed It Normal and Faulty Data

This publisher replays normal behavior and, after a minute, simulates a blocked impeller: current stays up while flow drops.

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

Start live_score.py, then replay.py. Scores stay below the threshold for the first minute and rise sharply once flow drops, because the flow-per-amp relationship breaks while current and pressure still look ordinary.

Explaining an Alert to a Technician

An anomaly score alone does not tell anyone what to check. Isolation Forest has no built-in per-feature explanation, but a cheap approximation works well in practice: for each feature, compute how far the current value sits from its training distribution, for example as a robust z-score against the training median and interquartile range, and include the two or three most extreme features in the alert.

An alert that reads "score 0.71 above threshold 0.62; flow_per_amp unusually low, flowLpm_mean unusually low" points straight at the impeller or the suction side. That is far more actionable than "pump1 anomaly". If you need more rigorous attributions, SHAP values can be computed for tree ensembles, but they cost more compute and are rarely necessary for a first deployment.

It also helps to publish the feature values themselves on the anomaly topic at a low rate. When an engineer later investigates, they can see the inputs the model saw, not just its verdict.

Tuning Checklist for Isolation Forest on Sensor Data

1. Train on representative normal data. Include all regular operating modes, shifts and seasons you expect to see.
2. Exclude known faults from training, or keep contamination honest if you cannot.
3. Keep features physically meaningful. Ratios and differences often beat raw values.
4. Avoid timestamps and counters as features; they make every new reading look novel.
5. Set the threshold from validation data and revisit it after the first weeks of alerts.
6. Require persistence before alerting, as the streak counter above does.
7. Version the model and include the version in every published result.
8. Retrain on a schedule or when drift monitoring tells you the input distribution moved.

That last point is where anomaly detection meets MLOps for IoT: the detector is itself a model that needs monitoring.

Running It in Production

Move the service to an edge gateway or a server near your data, connect over TLS on port 8883 with credentials issued by your broker, and keep the topic structure. A private ConnectMQ broker gives you a live topic explorer to watch scores next to raw telemetry, and threshold alert rules in the portal can flag a numeric score above a limit as a simple second safety net. If you prefer not to operate brokers at all, a hosted MQTT broker like ConnectMQ handles certificates, authentication and uptime, while your model stays your own code. For prototyping without an account, the broker.connectmq.com public area is enough.

For the surrounding architecture, see IoT anomaly detection methods; for a lighter single-signal version of this service, see real-time anomaly detection in Python. The wider context is in the AI and IoT guide.

Get Started with ConnectMQ

Create a broker on the free Serverless plan at app.connectmq.com, no credit card required, and replace the public host in live_score.py with your own endpoint and credentials. Your Isolation Forest results will appear in the topic explorer the moment the first window fills.

Frequently Asked Questions

How much data do I need to train Isolation Forest on sensor data?

A few thousand feature windows covering all normal operating modes is usually enough to start. Coverage of modes matters more than sheer volume.

What contamination value should I use?

Use your best estimate of how much of the training data is abnormal, often well under one percent for curated normal data. Because the live service uses its own threshold, contamination mainly affects the default predict output.

Can Isolation Forest run on a microcontroller?

The trained trees are small, but feature computation and scikit-learn itself need a Linux-class device. Run it on a gateway or Raspberry Pi and keep microcontrollers on simple thresholds.

Why does my model flag everything after a process change?

Isolation Forest scores relative to its training data, so a new product or setpoint looks abnormal. Retrain with data from the new mode, or train separate models per mode.