Previsão de Séries Temporais em IoT com Python e MQTT

A previsão de séries temporais em IoT estima os valores futuros de um sinal de sensor, como temperatura, nível de tanque, vazão ou potência, a partir do seu próprio histórico e de entradas relacionadas. Neste tutorial você vai capturar dados de sensores de um broker MQTT com Python, reamostrá-los em uma série limpa, comparar um modelo estatístico com um modelo de machine learning usando um backtest adequado e publicar a previsão de volta no MQTT para que dashboards e controladores possam usá-la.

Por que prever dados de sensores

Previsões transformam monitoramento em planejamento. Uma previsão da temperatura de uma câmara fria diz se o compressor vai dar conta durante o pico da tarde. Uma previsão do nível de um tanque diz quando pedir a próxima entrega. Uma previsão da demanda de energia diz quando deslocar cargas. Em todos os casos, o valor está em agir antes que a medição atinja um limite.

Previsão é diferente de detecção de anomalias. A detecção de anomalias pergunta se o valor atual é incomum; a previsão pergunta qual será o valor mais tarde. As duas costumam trabalhar juntas: uma grande diferença entre previsão e medição já é, por si só, um sinal útil de anomalia, como descrito em detecção de anomalias em IoT.

O que torna a previsão de séries temporais em IoT complicada

Dados de sensores são mais bagunçados que os conjuntos de dados arrumados da maioria dos tutoriais de previsão:

A maior parte do trabalho em um projeto de previsão é preparar a série; o modelo é a parte menor. O artigo sobre dados de séries temporais industriais cobre práticas de amostragem e de timestamp que facilitam muito essa preparação.

Escolhendo horizonte, resolução e entradas

Antes de escrever código, defina três decisões, porque elas moldam tudo o que vem depois.

Horizonte. Até que ponto à frente a decisão precisa olhar? Um controlador de refrigeração pode precisar de 30 minutos; um planejador de entregas, de vários dias. O erro cresce com o horizonte, então não preveja além do que a decisão exige. Se você precisa de horizontes curtos e longos, muitas vezes é melhor treinar modelos separados do que esticar um só.

Resolução. O passo da previsão deve corresponder à frequência com que a decisão pode mudar. Prever o nível de um tanque a cada segundo não faz sentido se o caminhão de entrega é agendado uma vez por dia. Uma resolução mais grossa também suaviza o ruído do sensor e reduz o número de passos que o modelo precisa prever. Para a maioria dos sinais de edificações e de processo, de 5 a 60 minutos é uma faixa sensata.

Entradas. Decida quais fatores externos você consegue conhecer com antecedência. Features de calendário estão sempre disponíveis. Previsões do tempo, planos de produção e agendas de ocupação frequentemente também. Valores medidos de outros sensores só estão disponíveis até o presente, então só podem ser usados como entradas defasadas. Anote essa lista; ela é o contrato que evita vazamento de dados mais adiante no projeto.

Passo 1: capture dados de sensores do MQTT

Para um tutorial autocontido, este script simula um sensor de temperatura com ciclo diário e ruído, publica os dados no servidor MQTT público, próprio para testes, e os grava de volta em um arquivo CSV com um assinante. Os "minutos" simulados são comprimidos para que a execução termine rápido; os timestamps no payload carregam o tempo simulado.

# capture.py
import csv
import json
import math
import random
import threading
import time
import uuid

import paho.mqtt.client as mqtt

BROKER = "broker.connectmq.com"
TOPIC = f"connectmq/demo/{uuid.uuid4().hex[:8]}/coldroom1/temperature"
START = 1767225600  # 2026-01-01T00:00:00Z
STEPS = 14 * 24 * 4  # 14 dias com resolução de 15 minutos
received = []


def on_connect(client, userdata, flags, reason_code, properties):
    client.subscribe(TOPIC, qos=1)


def on_message(client, userdata, msg):
    p = json.loads(msg.payload)
    received.append((p["ts"], p["value"]))


def simulate(pub):
    for i in range(STEPS):
        t = START + i * 900
        hour = (t % 86400) / 3600
        weekend = time.gmtime(t).tm_wday >= 5
        value = 4.0 + 1.2 * math.sin((hour - 9) / 24 * 2 * math.pi) + (-0.4 if weekend else 0)
        value += random.gauss(0, 0.15)
        if random.random() < 0.02:
            continue  # simula uma leitura perdida
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # respeita o limite de 10 msg/s por cliente do broker público


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-cap-" + uuid.uuid4().hex[:8])
client.on_connect = on_connect
client.on_message = on_message
client.connect(BROKER, 1883, keepalive=60)
client.loop_start()
time.sleep(1)
simulate(client)
time.sleep(3)
client.loop_stop()

with open("coldroom.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["ts", "value"])
    w.writerows(sorted(received))
print("rows:", len(received), "topic:", TOPIC)

Em produção, os mesmos dados viriam do seu historiador ou de um assinante conectado ao seu próprio broker ConnectMQ com YOUR_USERNAME, YOUR_PASSWORD e YOUR_CLIENT_ID via TLS na porta 8883.

Passo 2: prepare a série

Carregue o CSV, converta os timestamps, reamostre para uma grade fixa de 15 minutos e interpole apenas as lacunas curtas. Lacunas mais longas devem continuar vazias para não criar um histórico falso.

# prepare.py
import pandas as pd

df = pd.read_csv("coldroom.csv")
df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True)
s = df.set_index("ts")["value"].resample("15min").mean()
s = s.interpolate(limit=4)  # preenche lacunas de até uma hora
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Passo 3: dois modelos para previsão de séries temporais em IoT

Um bom hábito é sempre comparar pelo menos três previsões: uma linha de base ingênua, um modelo estatístico e um modelo de machine learning. Se um modelo complexo não supera "o mesmo horário de ontem", não vale a pena implantá-lo.

# forecast.py
import numpy as np
import pandas as pd
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.metrics import mean_absolute_error
from statsmodels.tsa.holtwinters import ExponentialSmoothing

s = pd.read_parquet("coldroom.parquet")["value"].ffill()
DAY = 96          # passos de 15 minutos por dia
HORIZON = DAY     # prevê 24 horas à frente


def make_features(series):
    X = pd.DataFrame(index=series.index)
    X["lag_day"] = series.shift(DAY)
    X["lag_day2"] = series.shift(2 * DAY)
    X["roll_day_mean"] = series.shift(DAY).rolling(DAY).mean()
    X["hour"] = series.index.hour + series.index.minute / 60
    X["dow"] = series.index.dayofweek
    return X


results = {"naive": [], "holt_winters": [], "gbr": []}
# Backtest com origem móvel: prevê cada um dos últimos 4 dias a partir dos dados anteriores
for k in range(4, 0, -1):
    cut = len(s) - k * DAY
    train, test = s.iloc[:cut], s.iloc[cut:cut + HORIZON]

    naive = train.iloc[-DAY:].values
    results["naive"].append(mean_absolute_error(test, naive))

    hw = ExponentialSmoothing(train, trend=None, seasonal="add", seasonal_periods=DAY).fit()
    results["holt_winters"].append(mean_absolute_error(test, hw.forecast(HORIZON)))

    X = make_features(s)
    Xtr, ytr = X.iloc[:cut].dropna(), s.iloc[:cut][X.iloc[:cut].dropna().index]
    gbr = HistGradientBoostingRegressor(max_iter=300, learning_rate=0.05).fit(Xtr, ytr)
    results["gbr"].append(mean_absolute_error(test, gbr.predict(X.iloc[cut:cut + HORIZON])))

for name, maes in results.items():
    print(f"{name:13s} MAE {np.mean(maes):.3f}")

Instale as bibliotecas com pip install paho-mqtt pandas pyarrow scikit-learn statsmodels. O modelo de gradient boosting usa apenas defasagens de pelo menos um dia, de modo que toda feature é conhecida no momento da previsão para todo o horizonte de 24 horas. Esse é o erro mais comum em previsão com ML: usar uma defasagem de um passo no treinamento e depois não ter com o que alimentá-la ao prever mais à frente.

Passo 4: faça um backtest honesto

O loop acima é um backtest com origem móvel: cada fold treina apenas com dados anteriores ao seu ponto de corte e prevê o dia seguinte. Nunca embaralhe séries temporais antes de dividi-las. Reporte o erro na unidade do sinal (MAE em graus é mais fácil de discutir que um percentual perto de zero) e observe os erros por hora do dia, porque um modelo pode ser bom na média e ruim justamente no pico que interessa a você.

Checklist para uma avaliação confiável:

1. Use várias origens de previsão, não uma única divisão.
2. Compare com uma linha de base sazonal ingênua.
3. Garanta que cada feature esteja disponível no momento da previsão.
4. Inspecione os erros em torno de eventos como ciclos de degelo, feriados ou paradas.
5. Execute o backtest novamente após qualquer mudança no código de preparação.

Passo 5: publique as previsões de volta no MQTT

Uma previsão é mais útil onde as decisões acontecem, então publique-a ao lado da medição. Uma retained message faz com que qualquer novo assinante receba a previsão mais recente imediatamente.

# publish_forecast.py
import json
import time
import uuid

import pandas as pd
import paho.mqtt.client as mqtt
from statsmodels.tsa.holtwinters import ExponentialSmoothing

s = pd.read_parquet("coldroom.parquet")["value"].ffill()
model = ExponentialSmoothing(s, seasonal="add", seasonal_periods=96).fit()
fc = model.forecast(96)

payload = {
    "generated": int(time.time() * 1000),
    "model": "holt-winters-v1",
    "step_minutes": 15,
    "points": [{"ts": int(t.timestamp() * 1000), "value": round(v, 2)} for t, v in fc.items()],
}

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="fc-pub-" + uuid.uuid4().hex[:8])
client.connect("broker.connectmq.com", 1883, keepalive=60)
client.loop_start()
info = client.publish("connectmq/demo/forecast-demo/coldroom1/forecast", json.dumps(payload), qos=1, retain=True)
info.wait_for_publish()
client.loop_stop()
client.disconnect()
print(len(json.dumps(payload)), "bytes published")

Uma previsão de 96 pontos tem alguns kilobytes, bem dentro do limite de mensagem de 64 KB do broker público e do limite de 256 KB do plano gratuito Serverless. Troque o segmento de demonstração do topic por algo único antes de executar. A jusante, um dashboard plota forecast contra temperature, e um script de controle pode agir se a previsão ultrapassar um limite nas próximas horas.

Executando de forma contínua

Transforme os scripts em um pequeno serviço: retreine diariamente com as últimas semanas, gere a previsão a cada 15 minutos e publique. Registre o erro da previsão quando os valores reais chegarem e publique-o em um topic próprio, para perceber quando a precisão piora; esse é o início das práticas de monitoramento descritas em MLOps para IoT. Para uma versão específica do mesmo pipeline, veja previsão de carga de energia.

Comece com o ConnectMQ

Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e conecte seus sensores e seu serviço de previsão com as credenciais geradas. O ConnectMQ opera o broker MQTT gerenciado para você, e o explorador de topics em tempo real permite conferir medições e previsões lado a lado. Para o panorama mais amplo de como combinar modelos e dispositivos, leia o guia de IA e IoT e, quando escalar, um plano de broker MQTT na nuvem suporta mais conexões e mensagens maiores.

Perguntas frequentes

Qual é o melhor modelo para previsão de séries temporais em IoT?

Não existe um único melhor modelo. Comece com uma linha de base sazonal ingênua, depois compare um modelo de suavização exponencial ou ARIMA com um modelo de gradient boosting sobre features de defasagem e de calendário e fique com o que vencer em um backtest com origem móvel.

De quanto histórico eu preciso?

Pelo menos vários ciclos completos da sazonalidade mais longa que você quer capturar, por exemplo algumas semanas para padrões diários e vários meses para padrões semanais com feriados.

Como lidar com leituras de sensor ausentes?

Reamostre para um intervalo fixo, interpole apenas as lacunas curtas e deixe as quedas longas como ausentes. Modelos treinados com longos trechos de dados inventados aprendem padrões que nunca aconteceram.

Onde o modelo de previsão deve rodar?

Normalmente em um servidor ou na nuvem, como um cliente MQTT que assina as medições e publica as previsões. Modelos pequenos também podem rodar em gateways de borda, perto dos equipamentos.