Isolation Forest para Dados de Sensores: Treine Offline e Pontue Mensagens MQTT ao Vivo

Aplicar Isolation Forest a dados de sensores é uma das formas mais práticas de detectar anomalias que se escondem na combinação de vários sinais, como vibração, temperatura e corrente das bombas de uma estação de saneamento. Neste tutorial, você vai gerar dados históricos, criar features por janela, treinar um modelo com scikit-learn e rodá-lo em um serviço Python que pontua mensagens MQTT ao vivo e publica eventos de anomalia.

Por que o Isolation Forest funciona bem com dados de sensores

O Isolation Forest se baseia em uma ideia simples: anomalias são poucas e diferentes, então são mais fáceis de isolar. O algoritmo constrói muitas árvores aleatórias; cada árvore escolhe repetidamente uma feature e um valor de corte aleatórios. Pontos normais ficam em regiões densas e precisam de muitos cortes para serem separados. Pontos anômalos são isolados após apenas alguns. O comprimento médio do caminho entre as árvores se torna a pontuação de anomalia.

Esse design combina com sensores industriais por vários motivos:

• Não precisa de rótulos. Você treina com operação majoritariamente normal, algo que toda fábrica tem de sobra.
• Multivariado por natureza. Corrente alta é normal, vazão baixa é normal, mas corrente alta com vazão baixa é isolada rapidamente.
• Rápido. Treinar com centenas de milhares de linhas leva segundos, e a pontuação é barata o bastante para um gateway de borda.
• Poucos hiperparâmetros. O número de árvores, o tamanho da amostra e a contaminação esperada geralmente são tudo o que você ajusta.

Ele também tem limitações. Não entende o tempo por conta própria, então você precisa codificar o comportamento temporal nas features. E pontua em relação aos dados de treinamento, então um modelo treinado no verão pode sinalizar o inverno como anômalo.

Isolation Forest vs outros detectores

Se você ainda não testou estatísticas de sinal único, leia primeiro detecção de outliers em dados de sensores com Z-score, IQR e EWMA. O Isolation Forest mostra o seu valor quando esses métodos deixam passar relações entre sinais.

Pré-requisitos

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

Escolha um prefixo aleatório como connectmq/demo/if-5d21/.... Os exemplos usam o servidor MQTT público em broker.connectmq.com, onde todos os topics podem ser lidos por outras pessoas, então mantenha os dados sintéticos.

Passo 1: crie dados históricos de sensores

Em um projeto real, você exportaria o histórico do seu banco de dados. Para manter o tutorial autocontido, este script simula uma bomba com três sinais correlacionados: corrente do motor, vazão e pressão de saída.

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

Os três sinais variam juntos com a carga. Essa correlação é exatamente o que um limite por sinal não consegue enxergar.

Passo 2: crie features por janela

Pontuar leituras isoladas ignora o ruído e a dinâmica de curto prazo. Em vez disso, calcule features sobre uma janela deslizante curta: a média de cada sinal, o desvio-padrão e algumas razões com significado físico.

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

As razões importam mais do que parece. Um rotor obstruído reduz a vazão por ampère, e um vazamento reduz a pressão por vazão, mesmo quando todos os valores brutos estão dentro da faixa normal.

Passo 3: treine o 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}")

Dois detalhes merecem atenção. Primeiro, o score_samples do scikit-learn retorna valores menores para pontos mais anormais, então negá-lo produz uma pontuação intuitiva do tipo "maior é pior". Segundo, em vez de depender só do predict, armazene um limite explícito tirado da distribuição de pontuações do treinamento. Assim você pode ajustá-lo depois sem treinar de novo.

Passo 4: pontue mensagens MQTT ao vivo

O serviço abaixo assina a telemetria da bomba, guarda as dez últimas leituras, calcula as mesmas features e publica uma pontuação a cada leitura.

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

Passo 5: alimente-o com dados normais e com falhas

Este publicador reproduz o comportamento normal e, após um minuto, simula um rotor obstruído: a corrente continua alta enquanto a vazão cai.

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

Inicie live_score.py e depois replay.py. As pontuações ficam abaixo do limite no primeiro minuto e sobem bruscamente assim que a vazão cai, porque a relação vazão por ampère se rompe enquanto a corrente e a pressão ainda parecem normais.

Explicando um alerta para um técnico

Uma pontuação de anomalia sozinha não diz a ninguém o que verificar. O Isolation Forest não tem uma explicação nativa por feature, mas uma aproximação barata funciona bem na prática: para cada feature, calcule o quanto o valor atual está distante da sua distribuição no treinamento, por exemplo como um z-score robusto em relação à mediana e ao intervalo interquartil do treinamento, e inclua no alerta as duas ou três features mais extremas.

Um alerta que diz "pontuação 0,71 acima do limite 0,62; flow_per_amp anormalmente baixo, flowLpm_mean anormalmente baixo" aponta diretamente para o rotor ou para o lado da sucção. Isso é muito mais acionável do que "anomalia na pump1". Se você precisar de atribuições mais rigorosas, é possível calcular valores SHAP para ensembles de árvores, mas eles exigem mais processamento e raramente são necessários em uma primeira implantação.

Também ajuda publicar os próprios valores das features no topic de anomalia, a uma taxa baixa. Quando um engenheiro investigar depois, ele poderá ver as entradas que o modelo viu, e não apenas o veredito.

Checklist de ajuste do Isolation Forest para dados de sensores

1. Treine com dados normais representativos. Inclua todos os modos de operação regulares, turnos e estações que você espera encontrar.
2. Exclua falhas conhecidas do treinamento ou, se não for possível, mantenha a contaminação realista.
3. Mantenha as features com significado físico. Razões e diferenças costumam superar os valores brutos.
4. Evite timestamps e contadores como features; eles fazem toda nova leitura parecer inédita.
5. Defina o limite a partir de dados de validação e revise-o após as primeiras semanas de alertas.
6. Exija persistência antes de alertar, como faz o contador de sequência acima.
7. Versione o modelo e inclua a versão em todo resultado publicado.
8. Treine novamente de forma programada ou quando o monitoramento de drift indicar que a distribuição das entradas mudou.

Esse último ponto é onde a detecção de anomalias encontra o MLOps para IoT: o detector é, ele próprio, um modelo que precisa de monitoramento.

Rodando em produção

Mova o serviço para um gateway de borda ou um servidor próximo dos seus dados, conecte-se via TLS na porta 8883 com credenciais emitidas pelo seu broker e mantenha a estrutura de topics. Um broker privado do ConnectMQ oferece um explorador de topics ao vivo para acompanhar as pontuações ao lado da telemetria bruta, e regras de alerta por limite no portal podem sinalizar um score numérico acima de um limite como uma segunda rede de segurança simples. Se você preferir não operar brokers, um broker MQTT gerenciado como o ConnectMQ cuida de certificados, autenticação e disponibilidade, enquanto o seu modelo continua sendo código seu. Para prototipar sem conta, o servidor MQTT público em broker.connectmq.com é suficiente, desde que apenas com dados sintéticos.

Para a arquitetura ao redor, veja métodos de detecção de anomalias em IoT; para uma versão mais leve deste serviço, com um único sinal, veja detecção de anomalias em tempo real com Python. O contexto mais amplo está no guia de IA e IoT.

Comece a usar o ConnectMQ

Crie um broker no plano Serverless gratuito em app.connectmq.com, sem cartão de crédito, e substitua o host do servidor MQTT público em live_score.py pelo seu próprio endpoint e credenciais. Os resultados do seu Isolation Forest aparecerão no explorador de topics assim que a primeira janela for preenchida.

Perguntas frequentes

Quantos dados preciso para treinar um Isolation Forest com dados de sensores?

Alguns milhares de janelas de features cobrindo todos os modos normais de operação costumam bastar para começar. A cobertura dos modos importa mais do que o volume puro.

Qual valor de contaminação devo usar?

Use a sua melhor estimativa de quanto dos dados de treinamento é anormal, muitas vezes bem abaixo de um por cento para dados normais selecionados. Como o serviço ao vivo usa o próprio limite, a contaminação afeta principalmente a saída padrão do predict.

O Isolation Forest pode rodar em um microcontrolador?

As árvores treinadas são pequenas, mas o cálculo das features e o próprio scikit-learn precisam de um dispositivo da classe Linux. Rode-o em um gateway ou Raspberry Pi e deixe os microcontroladores com limites simples.

Por que o meu modelo sinaliza tudo depois de uma mudança de processo?

O Isolation Forest pontua em relação aos dados de treinamento, então um produto ou setpoint novo parece anormal. Treine novamente com dados do novo modo ou treine modelos separados por modo.