La detección de anomalías en tiempo real con Python puede ser sorprendentemente pequeña: un suscriptor, una ventana móvil por sensor y una llamada de publicación para las alertas. En este tutorial construirás un simulador de sensores y un detector en streaming que puntúa cada mensaje MQTT a medida que llega y publica eventos de anomalía con antirrebote, todo ello contra el servidor MQTT público gratuito, que puedes usar ahora mismo para pruebas y prototipos.
Qué vas a construir
El montaje tiene tres partes conectadas mediante topics:
- Un simulador publica lecturas de temperatura y vibración de unas cuantas máquinas una vez por segundo, con fallos inyectados de vez en cuando.
- Un detector se suscribe a toda la telemetría, mantiene estadísticas móviles por máquina y por señal, y calcula un z-score robusto para cada lectura.
- Un topic de alertas recibe un evento JSON cuando una señal permanece anómala durante varias lecturas consecutivas, y un evento de recuperación cuando vuelve a la normalidad.
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
Publicar una puntuación por cada lectura y alertas solo en los cambios mantiene los dashboards informativos sin inundar de notificaciones a las personas.
Requisitos previos
- Python 3.9 o posterior
- paho-mqtt 2.x
- Acceso de red a broker.connectmq.com en el puerto 1883 (u 8883 para TLS)
python -m venv venv
source venv/bin/activate # en Windows: venv\Scripts\activate
pip install "paho-mqtt>=2.0"
Elige un ID aleatorio para el prefijo de tus topics, por ejemplo a71c2e. El área pública de broker.connectmq.com es legible por cualquiera, así que usa solo datos sintéticos. El servidor público no es apto para datos de producción y tiene límites, incluidos 10 mensajes por segundo por cliente, muy por encima de lo que usa este tutorial.
Paso 1: el simulador de sensores
Guarda esto como simulator.py. Produce ruido realista, una deriva lenta de tipo diario y episodios de fallo aleatorios: un pico de temperatura, un escalón de vibración y un sensor bloqueado.
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)
Ejecútalo y déjalo en marcha. Puedes ver los mensajes con el cliente de mosquitto:
mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v
Paso 2: elegir un método de puntuación para la detección de anomalías en tiempo real con Python
Un z-score simple usa la media y la desviación típica, que a su vez se ven arrastradas por los valores atípicos que quieres detectar. Un z-score robusto usa en su lugar la mediana y la desviación absoluta mediana (MAD):
robust_z = 0.6745 * (x - median) / MAD
La constante escala la MAD para que la puntuación sea comparable a un z-score estándar en datos con distribución normal. Un umbral de alrededor de 3,5 es un punto de partida habitual. Dos comprobaciones adicionales detectan fallos que la puntuación pasa por alto:
- Detección de bloqueo. Si las últimas N lecturas son idénticas, probablemente el sensor esté congelado aunque el valor parezca normal.
- Calentamiento. No puntúes hasta que la ventana tenga suficientes muestras; de lo contrario, el ruido inicial genera falsas alertas.
Para comparar z-score, IQR y EWMA con sus ventajas e inconvenientes, consulta métodos de detección de outliers en datos de sensores.
Paso 3: el detector en streaming
Guarda esto como detector.py. Mantiene una ventana por máquina y señal, puntúa cada lectura, publica las puntuaciones y emite alertas con una regla de persistencia sencilla: se activa tras 3 anomalías consecutivas y se desactiva tras 5 lecturas normales consecutivas.
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 # lecturas guardadas por señal
WARMUP = 30 # lecturas antes de empezar a puntuar
THRESHOLD = 3.5 # límite del z-score robusto
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) # aprende solo de las lecturas normales
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()
Tras unos 30 segundos de calentamiento deberías ver líneas ALERT cada vez que el simulador inyecta un fallo, seguidas de un evento cleared unos segundos después de que termine.
Paso 4: observar las alertas
Suscríbete al topic de alertas desde cualquier lugar:
mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v
Las alertas se publican con QoS 1 para que sobrevivan a un breve corte de red; las puntuaciones usan QoS 0 porque la siguiente llega un segundo después de todos modos. El artículo niveles de QoS de MQTT explicados trata ese equilibrio en profundidad.
Paso 5: añadir un dashboard sencillo
Las alertas te dicen que algo ha ocurrido; una gráfica te dice cómo se ha desarrollado. Como el detector publica una puntuación por cada lectura, cualquier dashboard compatible con MQTT puede representar los valores en bruto y las puntuaciones uno al lado del otro sin tocar el detector. Algunas opciones que funcionan bien:
- Node-RED con los nodos de dashboard. Suscríbete a
+/telemetryy+/score, separa por máquina y alimenta gráficas de líneas. Un nodo switch puede convertir el topic de alertas en un indicador de estado en color por máquina. - Grafana con un plugin de fuente de datos MQTT. Útil cuando ya usas Grafana para otras métricas y quieres las puntuaciones de anomalías en las mismas pantallas.
- Una página de navegador con MQTT sobre WebSocket. Conéctate con MQTT.js a
wss://broker.connectmq.com:8084/mqtty dibuja las puntuaciones con cualquier biblioteca de gráficos.
Al mirar las gráficas, fíjate en cómo se comporta la puntuación justo antes y después de una alerta. Un detector limpio muestra una puntuación tranquila que salta bruscamente durante un fallo y vuelve a la calma después. Una puntuación ruidosa que ronda el umbral la mayor parte del tiempo indica que el umbral es demasiado estricto para esa señal o que la ventana es demasiado corta para estimar la línea base con fiabilidad.
En un broker privado de ConnectMQ, el explorador de topics en directo te ofrece la misma vista rápida sin construir nada, lo que resulta práctico durante el ajuste.
Por qué puntuar cada lectura y alertar pocas veces
Es tentador publicar solo las alertas para ahorrar ancho de banda. En la práctica, el flujo de puntuaciones es lo que te permite ajustar el detector más adelante. Puedes reproducir un día de puntuaciones, preguntarte cuántas alertas habría producido otro umbral y comparar detectores de forma justa. Las alertas, en cambio, van dirigidas a personas y deben ser escasas y significativas, por eso el detector aplica persistencia antes de activarlas. Mantener estas dos salidas en topics separados permite que cada consumidor se suscriba exactamente a lo que necesita.
Lista de ajuste
| Parámetro | Efecto si es demasiado bajo | Efecto si es demasiado alto | Punto de partida |
|---|---|---|---|
| WINDOW | La línea base salta con el ruido | Tarda en adaptarse a cambios legítimos | De 2 a 5 minutos de datos |
| THRESHOLD | Muchas falsas alertas | Pasa por alto fallos pequeños | 3,5 |
| RAISE_AFTER | Alertas por fallos aislados | Detección tardía | 3 lecturas |
| CLEAR_AFTER | Alertas que oscilan | Las alertas siguen abiertas demasiado tiempo | 5 lecturas |
| STUCK_AFTER | Marca señales que cambian despacio | Los sensores congelados pasan desapercibidos | Depende de la resolución |
Resolución de problemas
- No llega ningún mensaje. Comprueba que el simulador y el detector usan el mismo prefijo y que cada cliente tiene un client ID único y no vacío. Dos clientes con el mismo ID se desconectan mutuamente una y otra vez.
- Alertas constantes en una señal. Puede que la señal esté cuantizada y la MAD sea cero; entonces el pequeño valor mínimo dispara la puntuación. Aumenta ese valor mínimo hasta la resolución del sensor.
- Las alertas nunca se desactivan. La ventana dejó de aprender porque, tras un cambio de nivel legítimo, todos los valores cuentan como anómalos. Añade un recálculo periódico de la línea base o usa EWMA para señales que cambian despacio.
- El detector se queda atrás. Python gestiona miles de mensajes por segundo con este patrón, pero los modelos pesados en el callback bloquean el bucle. Traslada la puntuación a una cola de trabajo o ejecuta varios detectores con suscripciones compartidas de MQTT 5.
De la demo a producción
Pasa a un broker privado antes de que fluya ningún dato real. Cambia la conexión para usar TLS y credenciales:
client.username_pw_set("YOUR_USERNAME", "YOUR_PASSWORD")
client.tls_set() # usa el almacén de CA del sistema; ConnectMQ usa un certificado de Let's Encrypt
client.connect("YOUR_BROKER_HOST", 8883, keepalive=60)
Usa el client ID generado para esas credenciales, ya que ConnectMQ vincula los client IDs a las credenciales emitidas. Un broker MQTT gestionado también significa que el detector no es lo único que tienes que mantener en funcionamiento. Cuando las estadísticas de una sola señal dejen de ser suficientes, el siguiente paso es un modelo multivariante como Isolation Forest con datos de sensores. Para una visión más amplia de la arquitectura, consulta métodos y arquitectura de detección de anomalías IoT y la guía de IA e IoT.
Empieza con ConnectMQ
Crea un broker en el plan gratuito Serverless en app.connectmq.com, sin tarjeta de crédito, copia en el detector el usuario, la contraseña y el client ID generados, y observa las puntuaciones y las alertas en el explorador de topics en directo.
Preguntas frecuentes
¿Por qué usar un z-score robusto en lugar de uno normal?
La mediana y la MAD apenas se ven afectadas por los valores atípicos que intentas detectar, así que la línea base se mantiene estable durante un fallo. Una media y una desviación típica normales se desplazan hacia la anomalía y la ocultan.
¿Cuántos mensajes por segundo puede gestionar este detector en Python?
Las estadísticas móviles son baratas, así que el rendimiento suele estar limitado por la red y el análisis de JSON más que por la puntuación. Para volúmenes mayores, ejecuta varias instancias del detector con suscripciones compartidas en un broker que las admita.
¿Puedo ejecutar el detector en una Raspberry Pi?
Sí. El script usa solo la biblioteca estándar más paho-mqtt y funciona con holgura en una Raspberry Pi que actúe como gateway de borde.
¿Las alertas deben ser mensajes retained?
Los eventos de alerta en sí, no. Si quieres que los dashboards nuevos vean el estado actual, publica un topic de estado retenido aparte por máquina con la lista de alertas activas.
