Predicción de series temporales IoT con Python

La predicción de series temporales IoT anticipa los valores futuros de una señal de sensor, como la temperatura, el nivel de un tanque, el caudal o la potencia, a partir de su propio histórico y de entradas relacionadas. En este tutorial capturarás datos de sensores desde un broker MQTT con Python, los remuestrearás en una serie limpia, compararás un modelo estadístico con uno de machine learning mediante un backtest adecuado y publicarás la previsión de vuelta en MQTT para que dashboards y controladores puedan usarla.

Por qué prever los datos de sensores

Las previsiones convierten la monitorización en planificación. Una previsión de la temperatura de una cámara frigorífica te dice si el compresor dará abasto durante el pico de la tarde. Una previsión del nivel de un tanque te dice cuándo pedir la próxima entrega. Una previsión de la demanda energética te dice cuándo desplazar cargas. En todos los casos, el valor está en actuar antes de que la medición alcance un límite.

La previsión es distinta de la detección de anomalías. La detección de anomalías pregunta si el valor actual es inusual; la previsión pregunta cuál será el valor más adelante. Ambas suelen trabajar juntas: una gran diferencia entre la previsión y la medición es en sí misma una señal de anomalía útil, como se describe en detección de anomalías en IoT.

Qué complica la predicción de series temporales IoT

Los datos de sensores son más desordenados que los conjuntos de datos pulcros de la mayoría de los tutoriales de previsión:

Problema	Efecto	Solución
Timestamps irregulares	Los modelos esperan un espaciado uniforme	Remuestrear a un intervalo fijo
Huecos por cortes	Retardos ausentes, estacionalidad rota	Interpolar los huecos cortos, marcar los largos
Datos con informe por excepción	Pocos puntos cuando la señal es plana	Rellenar hacia delante antes de remuestrear
Varias estacionalidades	Los ciclos diario y semanal se solapan	Características de calendario o modelos multiestacionales
Factores externos	Meteorología, ocupación, planes de producción	Añadir características exógenas
Deriva y reinicios del sensor	Saltos de nivel que no son reales	Limpiar o marcar antes de entrenar

La mayor parte del trabajo de un proyecto de previsión consiste en preparar la serie; el modelo es la parte pequeña. El artículo sobre datos de series temporales industriales cubre las prácticas de muestreo y timestamps que facilitan mucho esta preparación.

Elegir horizonte, resolución y entradas

Antes de escribir código, fija tres decisiones, porque condicionan todo lo que viene después.

Horizonte. ¿Hasta dónde necesita mirar la decisión? Un controlador de refrigeración puede necesitar 30 minutos; un planificador de entregas, varios días. El error crece con el horizonte, así que no hagas previsiones más allá de lo que requiere la decisión. Si necesitas horizontes cortos y largos, suele ser mejor entrenar modelos separados que estirar uno solo.

Resolución. El paso de la previsión debe coincidir con la frecuencia con la que puede cambiar la decisión. Prever el nivel de un tanque cada segundo no tiene sentido si el camión de reparto se reserva una vez al día. Una resolución más gruesa también suaviza el ruido del sensor y reduce el número de pasos que el modelo debe predecir. Para la mayoría de las señales de edificios y procesos, entre 5 y 60 minutos es un rango sensato.

Entradas. Decide qué factores externos puedes conocer de antemano. Las características de calendario siempre están disponibles. Las previsiones meteorológicas, los planes de producción y los horarios de ocupación, a menudo. Los valores medidos de otros sensores solo están disponibles hasta el presente, así que solo pueden usarse como entradas con retardo. Deja esta lista por escrito; es el contrato que evita fugas de información más adelante en el proyecto.

Paso 1: Captura datos de sensores desde MQTT

Para que el tutorial sea autocontenido, este script simula un sensor de temperatura con un ciclo diario y ruido, lo publica en el servidor MQTT público gratuito (solo para pruebas, no para datos de producción) y lo registra de vuelta con un suscriptor en un archivo CSV. Los "minutos" simulados están comprimidos para que la ejecución termine rápido; los timestamps del payload llevan el tiempo 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 días con resolución 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 una lectura perdida
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # respeta el límite de 10 msg/s por cliente del 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)

En producción, los mismos datos vendrían de tu historian o de un suscriptor conectado a tu propio broker de ConnectMQ con YOUR_USERNAME, YOUR_PASSWORD y YOUR_CLIENT_ID sobre TLS en el puerto 8883.

Paso 2: Prepara la serie

Carga el CSV, convierte los timestamps, remuestrea a una rejilla fija de 15 minutos e interpola solo los huecos cortos. Los huecos más largos deben quedarse vacíos para no crear un 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)  # rellena huecos de hasta una hora
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Paso 3: Dos modelos para la predicción de series temporales IoT

Un buen hábito es comparar siempre al menos tres previsiones: una línea base ingenua, un modelo estadístico y un modelo de machine learning. Si un modelo complejo no supera a "a la misma hora ayer", no merece la pena desplegarlo.

# 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          # pasos de 15 minutos por día
HORIZON = DAY     # previsión a 24 horas vista


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 con origen móvil: prevé cada uno de los últimos 4 días con los datos 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}")

Instala las librerías con pip install paho-mqtt pandas pyarrow scikit-learn statsmodels. El modelo de gradient boosting solo usa retardos de al menos un día, de modo que todas las características se conocen en el momento de la previsión para todo el horizonte de 24 horas. Este es el error más común en la previsión con ML: usar un retardo de un paso en el entrenamiento y no tener después nada con qué alimentarlo al prever más lejos.

Paso 4: Haz un backtest honesto

El bucle anterior es un backtest con origen móvil: cada pliegue entrena solo con datos anteriores a su punto de corte y prevé el día siguiente. Nunca barajes una serie temporal antes de dividirla. Expresa el error en la unidad de la señal (un MAE en grados es más fácil de discutir que un porcentaje cerca de cero) y examina los errores por hora del día, porque un modelo puede ser bueno de media y malo justo en el pico que te importa.

Lista de comprobación para una evaluación fiable:

Usa varios orígenes de previsión, no una sola división.
Compara con una línea base estacional ingenua.
Asegúrate de que todas las características están disponibles en el momento de la previsión.
Revisa los errores en torno a eventos como ciclos de desescarche, festivos o paradas.
Repite el backtest tras cualquier cambio en el código de preparación.

Paso 5: Publica las previsiones de vuelta en MQTT

Una previsión es más útil donde se toman las decisiones, así que publícala junto a la medición. Un retained message garantiza que cualquier nuevo suscriptor recibe la última previsión de inmediato.

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

Una previsión de 96 puntos ocupa unos pocos kilobytes, muy por debajo del límite de mensaje de 64 KB del broker público y del límite de 256 KB del plan gratuito Serverless. Cambia el segmento de demo del topic por algo único antes de ejecutarlo. Aguas abajo, un dashboard dibuja forecast frente a temperature, y un script de control puede actuar si la previsión cruza un límite en las próximas horas.

Ejecutarlo de forma continua

Convierte los scripts en un pequeño servicio: reentrena a diario con las últimas semanas, prevé cada 15 minutos y publica. Registra el error de la previsión cuando lleguen los valores reales y publícalo como su propio topic para darte cuenta cuando la precisión se degrade; ese es el comienzo de las prácticas de monitorización descritas en MLOps para IoT. Para una versión del mismo pipeline específica de un dominio, consulta previsión de la carga energética.

Empieza con ConnectMQ

Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y conecta tus sensores y tu servicio de previsión con las credenciales generadas. ConnectMQ opera por ti el broker MQTT gestionado, y el explorador de topics en vivo te permite comprobar las mediciones y las previsiones lado a lado. Para una visión más amplia de cómo combinar modelos con dispositivos, lee la guía de IA e IoT, y cuando escales, un plan de broker MQTT en la nube admite más conexiones y mensajes más grandes.

Preguntas frecuentes

¿Qué modelo es el mejor para la predicción de series temporales IoT?

No hay un único modelo mejor. Empieza con una línea base estacional ingenua, compara después un modelo de suavizado exponencial o ARIMA con un modelo de gradient boosting sobre características de retardo y calendario, y quédate con el que gane en un backtest con origen móvil.

¿Cuánto histórico necesito?

Al menos varios ciclos completos de la estacionalidad más larga que quieras capturar; por ejemplo, unas semanas para patrones diarios y varios meses para patrones semanales con festivos.

¿Cómo gestiono las lecturas de sensor que faltan?

Remuestrea a un intervalo fijo, interpola solo los huecos cortos y deja vacíos los cortes largos. Los modelos entrenados con largos tramos de datos inventados aprenden patrones que nunca ocurrieron.

¿Dónde debe ejecutarse el modelo de previsión?

Normalmente en un servidor o en la nube, como un cliente MQTT que se suscribe a las mediciones y publica las previsiones. Los modelos pequeños también pueden ejecutarse en gateways edge cerca de los equipos.

Predicción de series temporales IoT con Python y MQTT