IoT Time Series Forecasting with Python and MQTT

IoT time series forecasting predicts the future values of a sensor signal, such as temperature, tank level, flow or power, from its own history and related inputs. In this tutorial you will capture sensor data from an MQTT broker with Python, resample it into a clean series, compare a statistical model with a machine learning model using a proper backtest, and publish the forecast back to MQTT so dashboards and controllers can use it.

Why Forecast Sensor Data

Forecasts turn monitoring into planning. A forecast of cold-room temperature tells you whether the compressor will keep up during the afternoon peak. A forecast of tank level tells you when to order the next delivery. A forecast of energy demand tells you when to shift loads. In each case the value lies in acting before the measurement reaches a limit.

Forecasting is different from anomaly detection. Anomaly detection asks whether the current value is unusual; forecasting asks what the value will be later. The two often work together: a large gap between forecast and measurement is itself a useful anomaly signal, as described in IoT anomaly detection.

What Makes IoT Time Series Forecasting Tricky

Sensor data is messier than the tidy datasets in most forecasting tutorials:

Most of the work in a forecasting project is preparing the series; the model is the smaller part. The industrial time series data article covers sampling and timestamp practices that make this preparation much easier.

Choosing Horizon, Resolution and Inputs

Before writing code, pin down three decisions, because they shape everything that follows.

Horizon. How far ahead does the decision need to look? A cooling controller may need 30 minutes, a delivery planner several days. Error grows with the horizon, so do not forecast further than the decision requires. If you need both short and long horizons, it is often better to train separate models than to stretch one.

Resolution. The forecast step should match how often the decision can change. Forecasting a tank level every second is pointless if the delivery truck is booked once a day. Coarser resolution also smooths sensor noise and reduces the number of steps the model must predict. For most building and process signals, 5 to 60 minutes is a sensible range.

Inputs. Decide which external drivers you can know in advance. Calendar features are always available. Weather forecasts, production plans and occupancy schedules often are. Measured values of other sensors are only available up to the present, so they can only be used as lagged inputs. Write this list down; it is the contract that prevents leakage later in the project.

Step 1: Capture Sensor Data from MQTT

For a self-contained tutorial, this script simulates a temperature sensor with a daily cycle and noise, publishes it to the public MQTT test broker, and records it back with a subscriber into a CSV file. Simulated "minutes" are compressed so the run finishes quickly; the timestamps in the payload carry the simulated time.

# 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 days at 15-minute resolution
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  # simulate a dropped reading
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # respect the public broker's 10 msg/s per client limit


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)

In production the same data would come from your historian, or from a subscriber connected to your own ConnectMQ broker with YOUR_USERNAME, YOUR_PASSWORD and YOUR_CLIENT_ID over TLS on port 8883.

Step 2: Prepare the Series

Load the CSV, convert timestamps, resample to a fixed 15-minute grid, and interpolate short gaps only. Longer gaps should stay missing so they do not create fake history.

# 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)  # fill gaps up to one hour
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Step 3: Two Models for IoT Time Series Forecasting

A good habit is to always compare at least three forecasts: a naive baseline, a statistical model and a machine learning model. If a complex model cannot beat "same time yesterday", it is not worth deploying.

# 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          # 15-minute steps per day
HORIZON = DAY     # forecast 24 hours ahead


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": []}
# Rolling-origin backtest: forecast each of the last 4 days from the data before it
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}")

Install the libraries with pip install paho-mqtt pandas pyarrow scikit-learn statsmodels. The gradient boosting model only uses lags of at least one day, so every feature is known at forecast time for the whole 24-hour horizon. This is the most common mistake in ML forecasting: using a one-step lag in training and then having nothing to feed it when forecasting further ahead.

Step 4: Backtest Honestly

The loop above is a rolling-origin backtest: each fold trains only on data before its cut-off and forecasts the following day. Never shuffle time series before splitting. Report error in the signal's unit (MAE in degrees is easier to discuss than a percentage near zero), and look at errors by hour of day, because a model can be good on average and bad exactly at the peak you care about.

Checklist for a trustworthy evaluation:

1. Use several forecast origins, not one split.
2. Compare with a naive seasonal baseline.
3. Make sure every feature is available at forecast time.
4. Inspect errors around events such as defrost cycles, holidays or shutdowns.
5. Re-run the backtest after any change to preparation code.

Step 5: Publish Forecasts Back to MQTT

A forecast is most useful where decisions happen, so publish it next to the measurement. A retained message means any new subscriber gets the latest forecast immediately.

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

A 96-point forecast is a few kilobytes, well within the public broker's 64 KB message limit and the 256 KB limit of the free Serverless plan. Change the demo topic segment to something unique before running it. Downstream, a dashboard plots forecast against temperature, and a control script can act if the forecast crosses a limit within the next hours.

Running It Continuously

Turn the scripts into a small service: retrain daily on the latest weeks, forecast every 15 minutes, and publish. Log the forecast error once actuals arrive and publish it as its own topic so you notice when accuracy degrades; that is the beginning of the monitoring practices described in IoT MLOps. For a domain-specific version of the same pipeline, see energy load forecasting.

Get Started with ConnectMQ

Create a free Serverless broker at app.connectmq.com, no credit card required, and connect your sensors and forecasting service with the generated credentials. ConnectMQ runs the hosted MQTT broker for you, and the live topic explorer lets you check measurements and forecasts side by side. For the bigger picture of combining models with devices, read the AI and IoT guide, and when you scale up, a cloud MQTT broker plan supports more connections and larger messages.

Frequently Asked Questions

Which model is best for IoT time series forecasting?

There is no single best model. Start with a naive seasonal baseline, then compare an exponential smoothing or ARIMA model with a gradient boosting model on lag and calendar features, and keep whichever wins in a rolling backtest.

How much history do I need?

At least several full cycles of the longest seasonality you want to capture, for example a few weeks for daily patterns and several months for weekly patterns with holidays.

How do I handle missing sensor readings?

Resample to a fixed interval, interpolate only short gaps, and leave long outages missing. Models trained on long stretches of invented data learn patterns that never happened.

Where should the forecasting model run?

Usually on a server or in the cloud, as an MQTT client that subscribes to measurements and publishes forecasts. Small models can also run on edge gateways close to the equipment.