Isolation Forest para datos de sensores: entrena offline y puntúa mensajes MQTT en directo

Aplicar Isolation Forest a datos de sensores es una de las formas más prácticas de detectar anomalías que se esconden en la combinación de varias señales. En este tutorial generarás datos históricos, diseñarás características por ventana, entrenarás un modelo con scikit-learn y lo ejecutarás en un servicio Python que puntúa mensajes MQTT en directo y publica eventos de anomalía.

Por qué Isolation Forest funciona bien con datos de sensores

Isolation Forest se basa en una idea sencilla: las anomalías son pocas y diferentes, así que son más fáciles de aislar. El algoritmo construye muchos árboles aleatorios; cada árbol elige repetidamente una característica aleatoria y un valor de corte aleatorio. Los puntos normales se encuentran en regiones densas y necesitan muchos cortes para quedar separados. Los puntos anómalos quedan aislados tras solo unos pocos. La longitud media del camino a lo largo de los árboles se convierte en la puntuación de anomalía.

Ese diseño encaja con los sensores industriales por varios motivos:

• No necesita etiquetas. Entrenas con un funcionamiento mayoritariamente normal, algo que abunda en cualquier planta.
• Multivariante por naturaleza. Una corriente alta es normal, un caudal bajo es normal, pero una corriente alta con un caudal bajo queda aislada rápidamente.
• Rápido. Entrenar con cientos de miles de filas lleva segundos, y la puntuación es lo bastante barata para un gateway de borde.
• Pocos hiperparámetros. El número de árboles, el tamaño de muestra y la contaminación esperada suelen ser todo lo que ajustas.

También tiene limitaciones. Por sí solo no entiende el tiempo, así que debes codificar el comportamiento temporal en las características. Y puntúa en relación con los datos de entrenamiento, de modo que un modelo entrenado en verano puede marcar el invierno como anómalo.

Isolation Forest frente a otros detectores

Si aún no has probado las estadísticas de una sola señal, lee primero detección de outliers en datos de sensores con Z-score, IQR y EWMA. Isolation Forest se gana su sitio cuando esos métodos no captan las relaciones entre señales.

Requisitos previos

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

Elige un prefijo aleatorio como connectmq/demo/if-5d21/.... Los ejemplos usan el servidor MQTT público gratuito broker.connectmq.com, donde todos los topics son legibles por otros: es para pruebas y prototipos, así que mantén los datos sintéticos.

Paso 1: crear datos históricos de sensores

En un proyecto real exportarías el historial desde tu base de datos. Para que el tutorial sea autocontenido, este script simula una bomba con tres señales correlacionadas: corriente del motor, caudal y presión de salida.

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

rng = np.random.default_rng(42)
n = 20_000  # una lectura por segundo, unas 5,5 horas
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())

Las tres señales se mueven juntas con la carga. Esa correlación es exactamente lo que un umbral por señal no puede ver.

Paso 2: diseñar características por ventana

Puntuar lecturas individuales ignora el ruido y la dinámica de corto plazo. En su lugar, calcula características sobre una ventana deslizante corta: la media de cada señal, su desviación típica y un par de cocientes con significado físico.

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

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


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)

Los cocientes importan más de lo que parece. Un impulsor obstruido reduce el caudal por amperio, y una fuga reduce la presión por caudal, aunque cada valor en bruto esté dentro de su rango normal.

Paso 3: entrenar el 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,   # proporción esperada de anomalías en los datos de entrenamiento
    random_state=0,
)
model.fit(X)

scores = -model.score_samples(X)        # más alto = más anómalo
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}")

Hay dos detalles que merecen atención. Primero, score_samples de scikit-learn devuelve valores más bajos para los puntos más anómalos, así que al negarlo se obtiene una puntuación intuitiva de "cuanto más alta, peor". Segundo, en lugar de depender solo de predict, guarda un umbral explícito tomado de la distribución de puntuaciones de entrenamiento. Así podrás ajustarlo más adelante sin reentrenar.

Paso 4: puntuar mensajes MQTT en directo

El siguiente servicio se suscribe a la telemetría de la bomba, guarda las diez últimas lecturas, calcula las mismas características y publica una puntuación con cada lectura.

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

Paso 5: alimentarlo con datos normales y con fallos

Este publicador reproduce un comportamiento normal y, al cabo de un minuto, simula un impulsor obstruido: la corriente se mantiene alta mientras el caudal cae.

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

Inicia live_score.py y después replay.py. Las puntuaciones se mantienen por debajo del umbral durante el primer minuto y suben bruscamente en cuanto cae el caudal, porque la relación caudal por amperio se rompe mientras la corriente y la presión siguen pareciendo normales.

Explicar una alerta a un técnico

Una puntuación de anomalía por sí sola no le dice a nadie qué revisar. Isolation Forest no incluye una explicación por característica, pero una aproximación barata funciona bien en la práctica: para cada característica, calcula lo lejos que está el valor actual de su distribución de entrenamiento, por ejemplo como un z-score robusto respecto a la mediana y el rango intercuartílico de entrenamiento, e incluye en la alerta las dos o tres características más extremas.

Una alerta que dice "puntuación 0,71 por encima del umbral 0,62; flow_per_amp inusualmente bajo, flowLpm_mean inusualmente bajo" apunta directamente al impulsor o al lado de aspiración. Eso es mucho más útil que "anomalía en pump1". Si necesitas atribuciones más rigurosas, se pueden calcular valores SHAP para conjuntos de árboles, pero requieren más cálculo y rara vez son necesarios en un primer despliegue.

También ayuda publicar los propios valores de las características en el topic de anomalías a una frecuencia baja. Cuando un ingeniero investigue más tarde, podrá ver las entradas que vio el modelo, no solo su veredicto.

Lista de ajuste de Isolation Forest para datos de sensores

1. Entrena con datos normales representativos. Incluye todos los modos de funcionamiento habituales, turnos y estaciones que esperas ver.
2. Excluye los fallos conocidos del entrenamiento o, si no puedes, mantén una contaminación realista.
3. Mantén características con significado físico. Los cocientes y las diferencias suelen superar a los valores en bruto.
4. Evita marcas de tiempo y contadores como características; hacen que cada lectura nueva parezca novedosa.
5. Fija el umbral a partir de datos de validación y revísalo tras las primeras semanas de alertas.
6. Exige persistencia antes de alertar, como hace el contador de rachas anterior.
7. Versiona el modelo e incluye la versión en cada resultado publicado.
8. Reentrena periódicamente o cuando la monitorización de deriva indique que la distribución de entrada ha cambiado.

Este último punto es donde la detección de anomalías se encuentra con MLOps para IoT: el detector es en sí mismo un modelo que necesita monitorización.

Ponerlo en producción

Traslada el servicio a un gateway de borde o a un servidor cercano a tus datos, conéctate por TLS en el puerto 8883 con las credenciales emitidas por tu broker y mantén la estructura de topics. Un broker privado de ConnectMQ te ofrece un explorador de topics en directo para ver las puntuaciones junto a la telemetría en bruto, y las reglas de alerta por umbral del portal pueden señalar un score numérico por encima de un límite como una segunda red de seguridad sencilla. Si prefieres no operar brokers en absoluto, un broker MQTT gestionado como ConnectMQ se encarga de los certificados, la autenticación y la disponibilidad, mientras tu modelo sigue siendo tu propio código. Para prototipar sin cuenta, basta con el servidor público broker.connectmq.com.

Para la arquitectura que lo rodea, consulta métodos de detección de anomalías IoT; para una versión más ligera de este servicio con una sola señal, consulta detección de anomalías en tiempo real con Python. El contexto general está en 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, y sustituye el host público de live_score.py por tu propio endpoint y tus credenciales. Los resultados de tu Isolation Forest aparecerán en el explorador de topics en cuanto se llene la primera ventana.

Preguntas frecuentes

¿Cuántos datos necesito para entrenar Isolation Forest con datos de sensores?

Para empezar, suelen bastar unos pocos miles de ventanas de características que cubran todos los modos de funcionamiento normales. La cobertura de los modos importa más que el volumen puro.

¿Qué valor de contaminación debo usar?

Usa tu mejor estimación de qué proporción de los datos de entrenamiento es anómala, a menudo muy por debajo del uno por ciento en datos normales depurados. Como el servicio en directo usa su propio umbral, la contaminación afecta sobre todo a la salida por defecto de predict.

¿Puede Isolation Forest ejecutarse en un microcontrolador?

Los árboles entrenados son pequeños, pero el cálculo de características y el propio scikit-learn necesitan un dispositivo de clase Linux. Ejecútalo en un gateway o una Raspberry Pi y deja los microcontroladores para umbrales sencillos.

¿Por qué mi modelo lo marca todo tras un cambio de proceso?

Isolation Forest puntúa en relación con sus datos de entrenamiento, así que un producto o una consigna nuevos parecen anómalos. Reentrena con datos del nuevo modo o entrena modelos separados por modo.