Real-time anomaly detection in Python can be surprisingly small: one subscriber, a rolling window per sensor and a publish call for alerts. In this tutorial you will build a sensor simulator and a streaming detector that scores every MQTT message as it arrives and publishes debounced anomaly events, all against a public test broker you can use right now.
What You Will Build
The setup has three parts connected by topics:
- A simulator publishes temperature and vibration readings for a few machines once per second, with occasional injected faults.
- A detector subscribes to all telemetry, keeps rolling statistics per machine and per signal, and computes a robust z-score for every reading.
- An alerts topic receives a JSON event when a signal stays anomalous for several consecutive readings, and a recovery event when it returns to normal.
connectmq/demo/<id>/plant/<machine>/telemetry simulator -> detector
connectmq/demo/<id>/plant/<machine>/score detector, every reading
connectmq/demo/<id>/plant/alerts detector, on state change
Publishing a score for every reading and alerts only on changes keeps dashboards informative without flooding people with notifications.
Prerequisites
- Python 3.9 or newer
- paho-mqtt 2.x
- Network access to broker.connectmq.com on port 1883 (or 8883 for TLS)
python -m venv venv
source venv/bin/activate # on Windows: venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Pick a random ID for your topic prefix, for example a71c2e. The public area of broker.connectmq.com is readable by everyone, so use synthetic data only. The public MQTT test broker guide lists its limits, including 10 messages per second per client, which this tutorial stays well under.
Step 1: The Sensor Simulator
Save this as simulator.py. It produces realistic noise, a slow daily-like drift, and random fault episodes: a temperature spike, a vibration step and a stuck sensor.
import json
import math
import random
import time
import uuid
import paho.mqtt.client as mqtt
PREFIX = "connectmq/demo/a71c2e/plant"
MACHINES = ["press1", "press2", "pump1"]
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"sim-{uuid.uuid4().hex[:8]}")
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
state = {m: {"fault": None, "left": 0, "stuck": None} for m in MACHINES}
t = 0
while True:
for m in MACHINES:
s = state[m]
if s["left"] == 0 and random.random() < 0.01:
s["fault"] = random.choice(["spike", "vibration", "stuck"])
s["left"] = random.randint(5, 15)
temp = 60 + 3 * math.sin(t / 300) + random.gauss(0, 0.4)
vib = 2.0 + random.gauss(0, 0.08)
if s["left"] > 0:
if s["fault"] == "spike":
temp += 8
elif s["fault"] == "vibration":
vib += 0.9
elif s["fault"] == "stuck":
s["stuck"] = s["stuck"] or round(temp, 2)
temp = s["stuck"]
s["left"] -= 1
if s["left"] == 0:
s["fault"], s["stuck"] = None, None
payload = {"ts": time.time(), "tempC": round(temp, 2), "vibMm": round(vib, 3)}
client.publish(f"{PREFIX}/{m}/telemetry", json.dumps(payload), qos=0)
t += 1
time.sleep(1)
Run it and leave it going. You can watch the messages with the mosquitto client:
mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v
Step 2: Choosing a Scoring Method for Real-Time Anomaly Detection in Python
A plain z-score uses the mean and standard deviation, which are themselves pulled around by the outliers you want to catch. A robust z-score uses the median and the median absolute deviation (MAD) instead:
robust_z = 0.6745 * (x - median) / MAD
The constant scales MAD so that the score is comparable to a standard z-score for normally distributed data. A threshold around 3.5 is a common starting point. Two additional checks catch failures the score misses:
- Stuck detection. If the last N readings are identical, the sensor is probably frozen even though the value looks normal.
- Warm-up. Do not score until the window has enough samples, otherwise early noise produces false alerts.
For a comparison of z-score, IQR and EWMA with their trade-offs, see sensor data outlier detection methods.
Step 3: The Streaming Detector
Save this as detector.py. It keeps a window per machine and signal, scores each reading, publishes scores and emits alerts with a simple persistence rule: raise after 3 consecutive anomalies, clear after 5 consecutive normal readings.
import json
import statistics
import time
import uuid
from collections import defaultdict, deque
import paho.mqtt.client as mqtt
PREFIX = "connectmq/demo/a71c2e/plant"
WINDOW = 120 # readings kept per signal
WARMUP = 30 # readings before scoring starts
THRESHOLD = 3.5 # robust z-score limit
RAISE_AFTER = 3
CLEAR_AFTER = 5
STUCK_AFTER = 8
windows = defaultdict(lambda: deque(maxlen=WINDOW))
recent = defaultdict(lambda: deque(maxlen=STUCK_AFTER))
streak = defaultdict(int)
calm = defaultdict(int)
active = set()
def robust_z(window, x):
med = statistics.median(window)
mad = statistics.median(abs(v - med) for v in window) or 1e-6
return 0.6745 * (x - med) / mad
def evaluate(key, x):
w = windows[key]
recent[key].append(x)
stuck = len(recent[key]) == STUCK_AFTER and len(set(recent[key])) == 1
if len(w) < WARMUP:
w.append(x)
return 0.0, False
z = robust_z(w, x)
anomalous = abs(z) > THRESHOLD or stuck
if not anomalous:
w.append(x) # learn only from normal readings
return z, anomalous
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):
machine = msg.topic.split("/")[-2]
try:
data = json.loads(msg.payload)
except json.JSONDecodeError:
return
scores = {}
for signal in ("tempC", "vibMm"):
if signal not in data:
continue
key = (machine, signal)
z, anomalous = evaluate(key, data[signal])
scores[signal] = round(z, 2)
if anomalous:
streak[key] += 1
calm[key] = 0
else:
calm[key] += 1
streak[key] = 0
if key not in active and streak[key] >= RAISE_AFTER:
active.add(key)
publish_alert(client, machine, signal, "raised", z, data[signal])
elif key in active and calm[key] >= CLEAR_AFTER:
active.discard(key)
publish_alert(client, machine, signal, "cleared", z, data[signal])
client.publish(f"{PREFIX}/{machine}/score", json.dumps(scores), qos=0)
def publish_alert(client, machine, signal, state, z, value):
event = {"ts": time.time(), "machine": machine, "signal": signal,
"state": state, "z": round(z, 2), "value": value,
"detector": "robust-z-v1", "threshold": THRESHOLD}
client.publish(f"{PREFIX}/alerts", json.dumps(event), qos=1)
print("ALERT", event)
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
client_id=f"detector-{uuid.uuid4().hex[:8]}")
client.on_connect = on_connect
client.on_message = on_message
client.reconnect_delay_set(min_delay=1, max_delay=30)
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_forever()
After about 30 seconds of warm-up you should see ALERT lines whenever the simulator injects a fault, followed by a cleared event a few seconds after it ends.
Step 4: Watching Alerts
Subscribe to the alerts topic from anywhere:
mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v
Alerts are published with QoS 1 so they survive a brief network hiccup; scores use QoS 0 because the next one arrives a second later anyway. The MQTT QoS levels explained article covers that trade-off in depth.
Step 5: Adding a Simple Dashboard
Alerts tell you something happened; a chart tells you how it developed. Because the detector publishes a score for every reading, any MQTT-capable dashboard can plot raw values and scores side by side without touching the detector. A few options that work well:
- Node-RED with the dashboard nodes. Subscribe to
+/telemetryand+/score, split by machine, and feed line charts. A switch node can turn the alerts topic into a colored status indicator per machine. - Grafana with an MQTT data source plugin. Useful when you already use Grafana for other metrics and want anomaly scores on the same screens.
- A browser page using MQTT over WebSocket. Connect with MQTT.js to
wss://broker.connectmq.com:8084/mqttand draw the scores with any charting library.
When you look at the charts, pay attention to how the score behaves just before and after an alert. A clean detector shows a quiet score that jumps sharply during a fault and returns to quiet afterwards. A noisy score that hovers near the threshold most of the time means the threshold is too tight for that signal, or the window is too short to estimate the baseline reliably.
On a private ConnectMQ broker the live topic explorer gives you the same quick view without building anything, which is handy while tuning.
Why Score Every Reading and Alert Rarely
It is tempting to publish only alerts to save bandwidth. In practice, the score stream is what lets you tune the detector later. You can replay a day of scores, ask how many alerts a different threshold would have produced, and compare detectors fairly. Alerts, on the other hand, go to people and must be rare and meaningful, which is why the detector applies persistence before raising them. Keeping these two outputs on separate topics lets each consumer subscribe to exactly what it needs.
Tuning Checklist
| Parameter | Effect if too low | Effect if too high | Starting point |
|---|---|---|---|
| WINDOW | Baseline jumps with noise | Slow to adapt to legitimate changes | 2 to 5 minutes of data |
| THRESHOLD | Many false alerts | Misses small faults | 3.5 |
| RAISE_AFTER | Alerts on single glitches | Delayed detection | 3 readings |
| CLEAR_AFTER | Alert flapping | Alerts stay open too long | 5 readings |
| STUCK_AFTER | Flags slow-changing signals | Frozen sensors go unnoticed | Depends on resolution |
Troubleshooting
- No messages arrive. Check that the simulator and detector use the same prefix and that each client has a unique, non-empty client ID. Two clients with the same ID keep disconnecting each other.
- Constant alerts on one signal. The signal may be quantized so MAD is zero; the small floor value then explodes the score. Increase the floor to the sensor's resolution.
- Alerts never clear. The window stopped learning because every value counts as anomalous after a legitimate level shift. Add a periodic re-baseline or use EWMA for slow-changing signals.
- Detector falls behind. Python handles thousands of messages per second in this pattern, but heavy models in the callback block the loop. Move scoring to a worker queue, or run several detectors with MQTT 5 shared subscriptions.
From Demo to Production
Move to a private broker before any real data flows. Swap the connection for TLS and credentials:
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set() # uses the system CA store; ConnectMQ uses a Let's Encrypt certificate
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)
Use the client ID generated for those credentials, since ConnectMQ binds client IDs to issued credentials. A managed MQTT broker also means the detector is not the only thing you have to keep running. When single-signal statistics stop being enough, the next step is a multivariate model such as Isolation Forest on sensor data. For the bigger architectural picture, see IoT anomaly detection methods and architecture and 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, copy the generated username, password and client ID into the detector, and watch scores and alerts in the live topic explorer.
Frequently Asked Questions
Why use a robust z-score instead of a normal z-score?
The median and MAD are barely affected by the outliers you are trying to detect, so the baseline stays stable during a fault. A regular mean and standard deviation get pulled toward the anomaly and hide it.
How many messages per second can this Python detector handle?
The rolling statistics are cheap, so throughput is usually limited by network and JSON parsing rather than scoring. For higher volumes, run several detector instances with shared subscriptions on a broker that supports them.
Can I run the detector on a Raspberry Pi?
Yes. The script uses only the standard library plus paho-mqtt and runs comfortably on a Raspberry Pi acting as an edge gateway.
Should alerts be retained messages?
Not the alert events themselves. If you want new dashboards to see current state, publish a separate retained status topic per machine with the list of active alerts.
