Detecção de Anomalias em Tempo Real com Python e MQTT

A detecção de anomalias em tempo real com Python pode ser surpreendentemente pequena: um assinante, uma janela móvel por sensor e uma chamada de publish para os alertas. Neste tutorial, você vai construir um simulador de sensores e um detector em streaming que pontua cada mensagem MQTT assim que ela chega e publica eventos de anomalia com debounce, tudo no servidor MQTT público (gratuito), que você pode usar agora mesmo para testes e protótipos.

O que você vai construir

A configuração tem três partes conectadas por topics:

Um simulador publica leituras de temperatura e vibração de algumas máquinas (pense em duas prensas e uma bomba de uma planta de autopeças em Campinas) uma vez por segundo, com falhas injetadas ocasionalmente.
Um detector assina toda a telemetria, mantém estatísticas móveis por máquina e por sinal e calcula um z-score robusto para cada leitura.
Um topic de alertas recebe um evento JSON quando um sinal permanece anômalo por várias leituras consecutivas e um evento de recuperação quando ele volta ao 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

Publicar uma pontuação para cada leitura e alertas apenas nas mudanças mantém os dashboards informativos sem inundar as pessoas de notificações.

Pré-requisitos

Python 3.9 ou mais recente
paho-mqtt 2.x
Acesso de rede a broker.connectmq.com na porta 1883 (ou 8883 para TLS)

python -m venv venv
source venv/bin/activate        # on Windows: venv\Scripts\activate
pip install "paho-mqtt>=2.0"

Escolha um ID aleatório para o prefixo dos seus topics, por exemplo a71c2e. A área pública de broker.connectmq.com pode ser lida por qualquer pessoa, então use apenas dados sintéticos, nunca dados de produção. O servidor MQTT público tem limites, incluindo 10 mensagens por segundo por cliente, e este tutorial fica bem abaixo disso.

Passo 1: o simulador de sensores

Salve isto como simulator.py. Ele gera ruído realista, uma deriva lenta semelhante a um ciclo diário e episódios aleatórios de falha: um pico de temperatura, um degrau de vibração e um sensor travado.

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)

Execute-o e deixe rodando. Você pode acompanhar as mensagens com o cliente mosquitto:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-a71c2e -t "connectmq/demo/a71c2e/plant/#" -v

Passo 2: escolhendo um método de pontuação para detecção de anomalias em tempo real com Python

Um z-score simples usa a média e o desvio-padrão, que também são puxados pelos próprios outliers que você quer detectar. Um z-score robusto usa, em vez disso, a mediana e o desvio absoluto mediano (MAD):

robust_z = 0.6745 * (x - median) / MAD

A constante escala o MAD para que a pontuação seja comparável a um z-score padrão em dados com distribuição normal. Um limite em torno de 3,5 é um ponto de partida comum. Duas verificações adicionais capturam falhas que a pontuação deixa passar:

Detecção de travamento. Se as últimas N leituras forem idênticas, o sensor provavelmente está congelado, mesmo que o valor pareça normal.
Aquecimento. Não pontue até que a janela tenha amostras suficientes; caso contrário, o ruído inicial gera alertas falsos.

Para uma comparação entre z-score, IQR e EWMA com seus trade-offs, veja métodos de detecção de outliers em dados de sensores.

Passo 3: o detector em streaming

Salve isto como detector.py. Ele mantém uma janela por máquina e por sinal, pontua cada leitura, publica as pontuações e emite alertas com uma regra simples de persistência: dispara após 3 anomalias consecutivas e encerra após 5 leituras normais 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          # 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()

Depois de cerca de 30 segundos de aquecimento, você deve ver linhas ALERT sempre que o simulador injetar uma falha, seguidas de um evento cleared alguns segundos após o fim dela.

Passo 4: acompanhando os alertas

Assine o topic de alertas de qualquer lugar:

mosquitto_sub -h broker.connectmq.com -p 1883 -i alerts-a71c2e -t "connectmq/demo/a71c2e/plant/alerts" -v

Os alertas são publicados com QoS 1 para sobreviverem a uma breve instabilidade de rede; as pontuações usam QoS 0 porque a próxima chega um segundo depois de qualquer forma. O artigo níveis de QoS do MQTT explicados aborda esse trade-off em profundidade.

Passo 5: adicionando um dashboard simples

Os alertas dizem que algo aconteceu; um gráfico mostra como isso evoluiu. Como o detector publica uma pontuação para cada leitura, qualquer dashboard compatível com MQTT pode plotar valores brutos e pontuações lado a lado sem mexer no detector. Algumas opções que funcionam bem:

Node-RED com os nós de dashboard. Assine +/telemetry e +/score, separe por máquina e alimente gráficos de linha. Um nó switch pode transformar o topic de alertas em um indicador de status colorido por máquina.
Grafana com um plugin de fonte de dados MQTT. Útil quando você já usa o Grafana para outras métricas e quer as pontuações de anomalia nas mesmas telas.
Uma página no navegador usando MQTT sobre WebSocket. Conecte-se com MQTT.js a wss://broker.connectmq.com:8084/mqtt e desenhe as pontuações com qualquer biblioteca de gráficos.

Ao olhar os gráficos, preste atenção em como a pontuação se comporta logo antes e logo depois de um alerta. Um detector limpo mostra uma pontuação tranquila que salta bruscamente durante uma falha e volta a ficar tranquila depois. Uma pontuação ruidosa que fica rondando o limite a maior parte do tempo significa que o limite está apertado demais para aquele sinal ou que a janela é curta demais para estimar a linha de base de forma confiável.

Em um broker privado do ConnectMQ, o explorador de topics ao vivo oferece a mesma visão rápida sem construir nada, o que é prático durante o ajuste.

Por que pontuar cada leitura e alertar raramente

É tentador publicar apenas os alertas para economizar banda. Na prática, o fluxo de pontuações é o que permite ajustar o detector depois. Você pode reproduzir um dia de pontuações, perguntar quantos alertas um limite diferente teria gerado e comparar detectores de forma justa. Os alertas, por outro lado, vão para pessoas e precisam ser raros e significativos, e é por isso que o detector aplica persistência antes de dispará-los. Manter essas duas saídas em topics separados permite que cada consumidor assine exatamente o que precisa.

Checklist de ajuste

Parâmetro	Efeito se muito baixo	Efeito se muito alto	Ponto de partida
WINDOW	A linha de base salta com o ruído	Demora a se adaptar a mudanças legítimas	2 a 5 minutos de dados
THRESHOLD	Muitos alertas falsos	Deixa passar falhas pequenas	3,5
RAISE_AFTER	Alertas por falhas isoladas	Detecção atrasada	3 leituras
CLEAR_AFTER	Alertas oscilando	Alertas ficam abertos tempo demais	5 leituras
STUCK_AFTER	Sinaliza sinais que mudam devagar	Sensores congelados passam despercebidos	Depende da resolução

Solução de problemas

Nenhuma mensagem chega. Verifique se o simulador e o detector usam o mesmo prefixo e se cada cliente tem um client ID único e não vazio. Dois clientes com o mesmo ID ficam desconectando um ao outro.
Alertas constantes em um sinal. O sinal pode estar quantizado, de modo que o MAD é zero; o pequeno valor mínimo então faz a pontuação explodir. Aumente esse valor mínimo para a resolução do sensor.
Os alertas nunca se encerram. A janela parou de aprender porque, após uma mudança de nível legítima, todo valor conta como anômalo. Adicione uma redefinição periódica da linha de base ou use EWMA para sinais que mudam devagar.
O detector fica para trás. Python processa milhares de mensagens por segundo nesse padrão, mas modelos pesados no callback bloqueiam o loop. Mova a pontuação para uma fila de workers ou rode vários detectores com shared subscriptions do MQTT 5.

Da demonstração à produção

Migre para um broker privado antes que qualquer dado real comece a circular. Troque a conexão por TLS e credenciais:

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 o client ID gerado para essas credenciais, já que o ConnectMQ vincula os client IDs às credenciais emitidas. Um broker MQTT gerenciado também significa que o detector não é a única coisa que você precisa manter rodando. Quando as estatísticas de um único sinal deixarem de ser suficientes, o próximo passo é um modelo multivariado como o Isolation Forest em dados de sensores. Para o panorama arquitetural mais amplo, veja métodos e arquitetura de detecção de anomalias em IoT e o 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, copie o username, a senha e o client ID gerados para o detector e acompanhe pontuações e alertas no explorador de topics ao vivo.

Perguntas frequentes

Por que usar um z-score robusto em vez de um z-score normal?

A mediana e o MAD quase não são afetados pelos outliers que você está tentando detectar, então a linha de base permanece estável durante uma falha. A média e o desvio-padrão comuns são puxados em direção à anomalia e acabam escondendo-a.

Quantas mensagens por segundo este detector em Python consegue processar?

As estatísticas móveis são baratas, então a vazão geralmente é limitada pela rede e pelo parsing de JSON, e não pela pontuação. Para volumes maiores, rode várias instâncias do detector com shared subscriptions em um broker que as suporte.

Posso rodar o detector em um Raspberry Pi?

Sim. O script usa apenas a biblioteca padrão mais o paho-mqtt e roda com folga em um Raspberry Pi atuando como gateway de borda.

Os alertas devem ser retained messages?

Os eventos de alerta em si, não. Se você quiser que novos dashboards vejam o estado atual, publique um topic de status retido separado por máquina com a lista de alertas ativos.

Detecção de Anomalias em Tempo Real com Python em Fluxos de Sensores MQTT