A detecção de outliers em dados de sensores não precisa de machine learning para ser útil. Três técnicas clássicas, o Z-score, o intervalo interquartil (IQR) e a média móvel exponencialmente ponderada (EWMA), detectam a maior parte dos picos, quedas e derivas em fluxos de temperatura. Este artigo explica como cada uma funciona, onde ela falha e traz código Python que aplica as três a um feed MQTT ao vivo.
Outliers vs anomalias em fluxos de sensores
Um outlier é uma leitura que fica longe das demais. Uma anomalia é uma leitura ou um padrão que indica que algo incomum está acontecendo no mundo real. Os dois se sobrepõem, mas não são idênticos. Uma única leitura absurda causada por ruído elétrico é um outlier, e não uma anomalia de processo; uma subida lenta de dois graus ao longo de uma hora pode nunca gerar um outlier e ainda assim indicar um ventilador de resfriamento com defeito.
Por isso, a detecção de outliers em dados de sensores tem duas funções:
- Limpeza. Remover ou sinalizar picos e valores impossíveis antes que cheguem a dashboards, médias e modelos.
- Alerta antecipado. Destacar leituras que merecem atenção porque quebram o padrão recente.
Os métodos abaixo lidam com as duas, com pontos fortes diferentes. Para a arquitetura mais ampla, incluindo detectores de machine learning, veja métodos de detecção de anomalias em IoT.
Z-Score: distância em desvios-padrão
O Z-score mede quantos desvios-padrão um valor está distante da média:
z = (x - mean) / std
Uma leitura com Z-score absoluto acima de 3 é uma regra comum para outliers. Em um contexto de streaming, você calcula a média e o desvio-padrão sobre uma janela móvel de leituras recentes em vez de todo o histórico, para que a linha de base acompanhe mudanças lentas.
Pontos fortes: fácil de explicar, rápido e funciona bem quando o sinal é aproximadamente normal em torno de um nível estável.
Pontos fracos: a média e o desvio-padrão são, eles próprios, distorcidos pelos outliers. Uma rajada de picos infla o desvio-padrão e esconde o próximo pico. O método também pressupõe ruído simétrico, o que muitos sensores não têm.
Uma variante robusta substitui a média pela mediana e o desvio-padrão pelo desvio absoluto mediano (MAD). Essa versão é usada no tutorial de detecção de anomalias em tempo real com Python.
IQR: limites por quartis
O intervalo interquartil usa percentis em vez de médias:
Q1 = 25th percentile, Q3 = 75th percentile, IQR = Q3 - Q1
lower fence = Q1 - 1.5 * IQR
upper fence = Q3 + 1.5 * IQR
Tudo o que estiver fora desses limites é um outlier. Use 3 em vez de 1,5 para outliers "extremos".
Pontos fortes: robusto aos outliers que procura, não pressupõe normalidade e lida com dados assimétricos, como amplitudes de vibração ou consumo de corrente.
Pontos fracos: precisa de uma janela grande o bastante para quartis estáveis e, em sensores muito quantizados (um sensor de temperatura com resolução de 0,5 grau, por exemplo), o IQR pode colapsar para zero e sinalizar qualquer pequena mudança. Adicione uma largura mínima de limite igual à resolução do sensor.
EWMA: linha de base suavizada com limites de controle
A EWMA mantém um valor suavizado contínuo em que as leituras recentes pesam mais:
ewma = alpha * x + (1 - alpha) * ewma_previous
Junto com ela, você acompanha uma variância exponencialmente ponderada e sinaliza leituras que caem fora de ewma ± k * ewm_std. Um alpha pequeno (0,01 a 0,05) gera uma linha de base lenta e estável; um alpha maior acompanha as mudanças rapidamente.
Pontos fortes: memória constante (dois números por sinal), ideal para microcontroladores, adapta-se naturalmente a derivas lentas e é sensível a pequenas mudanças sustentadas quando usada como gráfico de controle.
Pontos fracos: como se adapta, uma falha lenta pode ser absorvida pela linha de base. Combine-a com um limite absoluto para valores que nunca podem ser ultrapassados.
Comparando os métodos
| Método | Memória por sinal | Robusto a outliers | Lida com assimetria | Detecta deriva lenta | Cabe em microcontrolador |
|---|---|---|---|---|---|
| Z-score móvel | Janela de N valores | Não | Não | Mal | Sim, com N pequeno |
| Z robusto (mediana/MAD) | Janela de N valores | Sim | Em parte | Mal | Sim, com N pequeno |
| Limites por IQR | Janela de N valores | Sim | Sim | Mal | Sim, com N pequeno |
| Limites de controle EWMA | Dois números | Em parte | Não | Sim, com alpha pequeno | Facilmente |
Nenhum método vence sozinho. Uma combinação prática para temperatura é: limites físicos rígidos, IQR para picos e EWMA com alpha pequeno para deriva.
Detecção de outliers em dados de sensores com Python
O código abaixo implementa os três detectores como pequenas classes com um método update, para que você possa rodá-los lado a lado no mesmo fluxo.
# detectors.py
import math
import statistics
from collections import deque
class RollingZ:
def __init__(self, window=60, k=3.0):
self.buf, self.k = deque(maxlen=window), k
def update(self, x):
flagged = False
if len(self.buf) >= 20:
mean = statistics.fmean(self.buf)
std = statistics.pstdev(self.buf) or 1e-9
flagged = abs(x - mean) / std > self.k
if not flagged:
self.buf.append(x)
return flagged
class IQRFence:
def __init__(self, window=60, k=1.5, min_width=0.2):
self.buf, self.k, self.min_width = deque(maxlen=window), k, min_width
def update(self, x):
flagged = False
if len(self.buf) >= 20:
q1, _, q3 = statistics.quantiles(self.buf, n=4)
iqr = max(q3 - q1, self.min_width)
flagged = x < q1 - self.k * iqr or x > q3 + self.k * iqr
if not flagged:
self.buf.append(x)
return flagged
class EWMA:
def __init__(self, alpha=0.05, k=4.0, warmup=20):
self.alpha, self.k, self.warmup = alpha, k, warmup
self.mean, self.var, self.n = None, 0.0, 0
def update(self, x):
self.n += 1
if self.mean is None:
self.mean = x
return False
diff = x - self.mean
std = math.sqrt(self.var) or 1e-9
flagged = self.n > self.warmup and abs(diff) > self.k * std
if not flagged:
self.mean += self.alpha * diff
self.var = (1 - self.alpha) * (self.var + self.alpha * diff * diff)
return flagged
Cada detector deixa de aprender com os valores que sinaliza, o que impede que um pico contamine a linha de base.
Aplicando a um fluxo MQTT de temperatura ao vivo
Para testar de ponta a ponta, publique temperaturas simuladas no servidor MQTT público (gratuito) em broker.connectmq.com e rode os detectores em um assinante. Substitua t9x4q2 pelo seu próprio ID aleatório; a área pública pode ser lida por qualquer pessoa, então use apenas dados sintéticos, nunca dados de produção.
# temp_sim.py
import json, random, time, uuid
import paho.mqtt.client as mqtt
TOPIC = "connectmq/demo/t9x4q2/lab/room1/temperature"
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"tsim-{uuid.uuid4().hex[:6]}")
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_start()
base = 21.0
while True:
base += random.gauss(0, 0.01) # slow wander
value = base + random.gauss(0, 0.15)
if random.random() < 0.02:
value += random.choice([-6, 7]) # occasional spike
c.publish(TOPIC, json.dumps({"tempC": round(value, 2), "ts": time.time()}))
time.sleep(1)
# temp_outliers.py
import json, uuid
import paho.mqtt.client as mqtt
from detectors import RollingZ, IQRFence, EWMA
TOPIC = "connectmq/demo/t9x4q2/lab/room1/temperature"
detectors = {"z": RollingZ(), "iqr": IQRFence(), "ewma": EWMA()}
def on_connect(client, userdata, flags, reason_code, properties):
client.subscribe(TOPIC)
def on_message(client, userdata, msg):
x = json.loads(msg.payload)["tempC"]
flags = {name: d.update(x) for name, d in detectors.items()}
if any(flags.values()):
print(f"{x:6.2f} flagged by: {[n for n, f in flags.items() if f]}")
client.publish(TOPIC.rsplit("/", 1)[0] + "/outliers",
json.dumps({"tempC": x, "flags": flags}), qos=1)
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"tout-{uuid.uuid4().hex[:6]}")
c.on_connect = on_connect
c.on_message = on_message
c.connect("broker.connectmq.com", 1883, keepalive=60)
c.loop_forever()
Rode os dois e observe quais detectores concordam. Normalmente, você verá os três pegarem os picos grandes, enquanto o Z-score às vezes deixa passar um pico que vem logo depois de outro, exatamente o ponto fraco descrito acima.
Detectar no dispositivo vs depois do broker
Você pode rodar esses detectores em dois lugares, e a escolha afeta o que você publica.
Rodar no nó sensor, por exemplo um ESP32, significa que o dispositivo pode anexar uma flag de outlier a cada mensagem ou publicar apenas quando algo interessante acontecer. Isso economiza banda e funciona mesmo com a rede fora do ar, mas os limites ficam embutidos no firmware e são mais difíceis de mudar.
Rodar depois do broker, em um assinante como o acima, mantém os dispositivos simples e permite mudar métodos e limites a qualquer momento, comparar vários detectores em paralelo e reprocessar o histórico. O custo é que toda leitura bruta precisa viajar até o broker.
Um meio-termo comum é um limite rígido e uma verificação de valor travado no dispositivo, que detectam sensores quebrados na hora, e os detectores estatísticos em um serviço assinante, onde são fáceis de ajustar. Com um broker ConnectMQ no meio, as duas saídas chegam a topics que os mesmos dashboards podem ler.
Checklist prática
- Aplique primeiro limites físicos rígidos (um sensor de ambiente marcando 150 graus é uma falha, não um outlier).
- Detecte valores travados separadamente, verificando leituras idênticas ao longo de um período.
- Defina o tamanho das janelas em tempo, não em amostras, se os intervalos de publicação variarem.
- Defina larguras mínimas de faixa iguais à resolução do sensor.
- Sinalize os outliers em vez de excluí-los e guarde o valor bruto para análise posterior.
- Exija duas ou três sinalizações consecutivas antes de alertar uma pessoa.
- Revise os limites após uma semana de dados reais.
Quando os sinais interagem, por exemplo corrente e vazão em uma bomba de uma estação de saneamento, esses métodos por sinal chegam ao seu limite, e um modelo multivariado como o Isolation Forest em dados de sensores passa a valer a pena.
Onde o broker se encaixa
Cada peça acima é um cliente MQTT comum: o sensor publica, o detector assina e republica. O broker MQTT apenas roteia as mensagens, o que significa que você pode adicionar, remover ou substituir detectores sem tocar nos dispositivos. Em implantações reais, um broker MQTT gerenciado privado como o ConnectMQ, um broker MQTT na nuvem, acrescenta TLS, credenciais e um explorador de topics ao vivo onde você pode ver leituras brutas e outliers sinalizados lado a lado. Para mais formas de combinar análises com dispositivos, veja 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, aponte os seus sensores para ele e rode o detector de outliers sobre os seus próprios dados em vez do simulador.
Perguntas frequentes
Qual método de detecção de outliers é melhor para sensores de temperatura?
Uma combinação funciona melhor: limites físicos, limites por IQR para picos e EWMA com um fator de suavização pequeno para derivas lentas. A temperatura muda devagar, então a EWMA é particularmente eficaz.
Que tamanho de janela devo usar para estatísticas móveis?
Cubra tempo suficiente para capturar o ruído normal, mas não tanto que mudanças legítimas levem uma eternidade para ser absorvidas. Para uma leitura por segundo, de um a cinco minutos é um ponto de partida razoável.
Os outliers devem ser removidos dos dados?
Sinalize-os em vez de excluí-los. Guarde o valor bruto com uma flag de outlier para poder auditar as decisões e treinar modelos novamente mais tarde.
Esses métodos podem rodar em um ESP32?
Sim. A EWMA precisa de apenas dois floats por sinal, e uma janela móvel de 30 a 60 valores cabe facilmente na memória do ESP32, então você pode sinalizar outliers antes de publicar.
