Python ve MQTT ile IoT Zaman Serisi Tahmini

IoT zaman serisi tahmini; sıcaklık, tank seviyesi, debi veya güç gibi bir sensör sinyalinin gelecekteki değerlerini, sinyalin kendi geçmişinden ve ilişkili girdilerden öngörür. Bu rehberde Python ile bir MQTT broker'dan sensör verisi toplayacak, bunu temiz bir seriye yeniden örnekleyecek, istatistiksel bir modeli bir makine öğrenmesi modeliyle düzgün bir geriye dönük testle (backtest) karşılaştıracak ve panoların ile kontrolörlerin kullanabilmesi için tahmini MQTT'ye geri yayınlayacaksınız.

Sensör Verisini Neden Tahmin Etmelisiniz?

Tahminler izlemeyi planlamaya dönüştürür. Soğuk oda sıcaklığı tahmini, kompresörün öğleden sonraki zirvede yetişip yetişemeyeceğini söyler. Tank seviyesi tahmini, bir sonraki teslimatın ne zaman sipariş edileceğini söyler. Enerji talebi tahmini, yüklerin ne zaman kaydırılacağını söyler. Her durumda değer, ölçüm bir sınıra ulaşmadan önce harekete geçmekte yatar.

Tahmin, anomali tespitinden farklıdır. Anomali tespiti güncel değerin olağandışı olup olmadığını sorar; tahmin ise değerin ileride ne olacağını sorar. İkisi çoğu zaman birlikte çalışır: tahmin ile ölçüm arasındaki büyük bir fark, IoT anomali tespiti yazısında anlatıldığı gibi başlı başına faydalı bir anomali sinyalidir.

IoT Zaman Serisi Tahminini Zorlaştıran Nedir?

Sensör verisi, çoğu tahmin eğitimindeki düzenli veri kümelerinden daha dağınıktır:

Bir tahmin projesindeki işin büyük kısmı seriyi hazırlamaktır; model daha küçük kısımdır. Endüstriyel zaman serisi verisi yazısı, bu hazırlığı çok daha kolay hale getiren örnekleme ve zaman damgası pratiklerini ele alıyor.

Ufuk, Çözünürlük ve Girdileri Seçmek

Kod yazmadan önce üç kararı netleştirin, çünkü sonraki her şeyi bunlar şekillendirir.

Ufuk. Karar ne kadar ileriye bakmalı? Bir soğutma kontrolörü 30 dakikaya, bir teslimat planlayıcısı birkaç güne ihtiyaç duyabilir. Hata ufukla birlikte büyür; bu yüzden kararın gerektirdiğinden daha uzağı tahmin etmeyin. Hem kısa hem uzun ufuklara ihtiyacınız varsa, tek bir modeli zorlamak yerine ayrı modeller eğitmek çoğu zaman daha iyidir.

Çözünürlük. Tahmin adımı, kararın ne sıklıkla değişebileceğiyle eşleşmelidir. Teslimat kamyonu günde bir kez rezerve ediliyorsa tank seviyesini her saniye tahmin etmek anlamsızdır. Daha kaba çözünürlük ayrıca sensör gürültüsünü yumuşatır ve modelin tahmin etmesi gereken adım sayısını azaltır. Çoğu bina ve proses sinyali için 5 ila 60 dakika makul bir aralıktır.

Girdiler. Hangi dış etkenleri önceden bilebileceğinize karar verin. Takvim öznitelikleri her zaman mevcuttur. Hava durumu tahminleri, üretim planları ve doluluk çizelgeleri çoğu zaman mevcuttur. Diğer sensörlerin ölçülen değerleri ise yalnızca şimdiye kadar mevcuttur; bu yüzden yalnızca gecikmeli girdi olarak kullanılabilirler. Bu listeyi yazın; projenin ilerleyen aşamalarında sızıntıyı önleyen sözleşme budur.

Adım 1: MQTT'den Sensör Verisi Toplayın

Kendi kendine yeten bir eğitim için bu betik, günlük döngüsü ve gürültüsü olan bir sıcaklık sensörünü simüle eder, onu herkese açık MQTT test broker'ına yayınlar ve bir aboneyle geri alıp bir CSV dosyasına kaydeder. Çalıştırmanın hızlı bitmesi için simüle "dakikalar" sıkıştırılmıştır; payload'daki zaman damgaları simüle zamanı taşır.

# 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  # 15 dakikalık çözünürlükte 14 gün
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  # kaybolan bir okumayı simüle et
        pub.publish(TOPIC, json.dumps({"ts": t * 1000, "value": round(value, 3)}), qos=0)
        time.sleep(0.12)  # herkese açık broker'ın istemci başına 10 mesaj/sn sınırına uy


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)

Üretimde aynı veri historian'ınızdan ya da 8883 portunda TLS üzerinden YOUR_USERNAME, YOUR_PASSWORD ve YOUR_CLIENT_ID ile kendi ConnectMQ broker'ınıza bağlanan bir aboneden gelir.

Adım 2: Seriyi Hazırlayın

CSV'yi yükleyin, zaman damgalarını dönüştürün, sabit 15 dakikalık bir ızgaraya yeniden örnekleyin ve yalnızca kısa boşlukları enterpole edin. Daha uzun boşluklar sahte geçmiş oluşturmamaları için eksik kalmalıdır.

# 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)  # bir saate kadar olan boşlukları doldur
print(s.isna().sum(), "missing after interpolation")
s.to_frame("value").to_parquet("coldroom.parquet")

Adım 3: IoT Zaman Serisi Tahmini İçin İki Model

İyi bir alışkanlık, her zaman en az üç tahmini karşılaştırmaktır: naif bir referans, istatistiksel bir model ve bir makine öğrenmesi modeli. Karmaşık bir model "dün aynı saat" tahminini geçemiyorsa devreye almaya değmez.

# 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          # günde 15 dakikalık adım sayısı
HORIZON = DAY     # 24 saat ilerisini tahmin et


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": []}
# Kayan başlangıçlı geriye dönük test: son 4 günün her birini kendisinden önceki veriyle tahmin et
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}")

Kütüphaneleri pip install paho-mqtt pandas pyarrow scikit-learn statsmodels ile kurun. Gradient boosting modeli yalnızca en az bir günlük gecikmeleri kullanır; böylece 24 saatlik ufkun tamamı için her öznitelik tahmin anında bilinir. ML tahminindeki en yaygın hata budur: eğitimde tek adımlık bir gecikme kullanıp daha ileriyi tahmin ederken onu besleyecek hiçbir şeyin olmaması.

Adım 4: Dürüst Bir Geriye Dönük Test Yapın

Yukarıdaki döngü kayan başlangıçlı (rolling-origin) bir geriye dönük testtir: her katman yalnızca kesme noktasından önceki veriyle eğitilir ve sonraki günü tahmin eder. Zaman serilerini bölmeden önce asla karıştırmayın. Hatayı sinyalin biriminde raporlayın (derece cinsinden MAE, sıfıra yakın bir yüzdeden daha kolay tartışılır) ve hataları günün saatine göre inceleyin; çünkü bir model ortalamada iyi, ama tam da önemsediğiniz zirvede kötü olabilir.

Güvenilir bir değerlendirme için kontrol listesi:

1. Tek bir bölme değil, birden fazla tahmin başlangıç noktası kullanın.
2. Naif mevsimsel bir referansla karşılaştırın.
3. Her özniteliğin tahmin anında mevcut olduğundan emin olun.
4. Buz çözme döngüleri, tatiller veya duruşlar gibi olayların etrafındaki hataları inceleyin.
5. Hazırlık kodundaki her değişiklikten sonra geriye dönük testi yeniden çalıştırın.

Adım 5: Tahminleri MQTT'ye Geri Yayınlayın

Bir tahmin, en çok kararların verildiği yerde işe yarar; bu yüzden onu ölçümün yanında yayınlayın. Retained message, her yeni abonenin en güncel tahmini hemen alması anlamına gelir.

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

96 noktalık bir tahmin birkaç kilobayttır; herkese açık broker'ın 64 KB mesaj sınırının ve ücretsiz Serverless planın 256 KB sınırının oldukça altındadır. Çalıştırmadan önce demo topic segmentini benzersiz bir şeyle değiştirin. Aşağı akışta bir pano forecast değerini temperature ile birlikte çizer; bir kontrol betiği de tahmin önümüzdeki saatlerde bir sınırı aşarsa harekete geçebilir.

Sürekli Çalıştırmak

Betikleri küçük bir servise dönüştürün: her gün son haftaların verisiyle yeniden eğitin, her 15 dakikada bir tahmin yapın ve yayınlayın. Gerçek değerler geldiğinde tahmin hatasını kaydedin ve doğruluk düştüğünde fark edebilmek için onu da ayrı bir topic olarak yayınlayın; bu, IoT MLOps yazısında anlatılan izleme pratiklerinin başlangıcıdır. Aynı veri hattının alana özgü bir sürümü için enerji yükü tahmini yazısına bakın.

ConnectMQ ile Başlayın

app.connectmq.com adresinde kredi kartı gerektirmeden ücretsiz bir Serverless broker oluşturun ve sensörlerinizi ile tahmin servisinizi oluşturulan kimlik bilgileriyle bağlayın. ConnectMQ sizin için barındırılan MQTT broker hizmetini çalıştırır; canlı topic gezgini de ölçümleri ve tahminleri yan yana kontrol etmenizi sağlar. Modelleri cihazlarla birleştirmenin daha geniş tablosu için yapay zeka ve IoT rehberini okuyun; ölçeklendiğinizde bir bulut MQTT broker planı daha fazla bağlantıyı ve daha büyük mesajları destekler.

Sık Sorulan Sorular

IoT zaman serisi tahmini için en iyi model hangisi?

Tek bir en iyi model yoktur. Naif mevsimsel bir referansla başlayın, ardından bir üstel düzleştirme veya ARIMA modelini gecikme ve takvim öznitelikleri üzerinde çalışan bir gradient boosting modeliyle karşılaştırın ve kayan bir geriye dönük testte kazananı tutun.

Ne kadar geçmiş veriye ihtiyacım var?

Yakalamak istediğiniz en uzun mevsimselliğin en az birkaç tam döngüsüne; örneğin günlük desenler için birkaç haftaya, tatilleri de içeren haftalık desenler için birkaç aya.

Eksik sensör okumalarını nasıl ele alırım?

Sabit bir aralığa yeniden örnekleyin, yalnızca kısa boşlukları enterpole edin ve uzun kesintileri eksik bırakın. Uzun süreli uydurulmuş veriyle eğitilen modeller hiç yaşanmamış desenleri öğrenir.

Tahmin modeli nerede çalışmalı?

Genellikle bir sunucuda veya bulutta, ölçümlere abone olup tahminleri yayınlayan bir MQTT istemcisi olarak. Küçük modeller ekipmana yakın edge gateway'lerde de çalışabilir.