REST API s knihovnou requests
Funkce prices() vrací JSON v2 pro zvolený den. Odpověď 404 znamená, že pro den zatím nejsou data, API nikdy nevrací jiný den. Časy intervalů jsou ISO 8601 s posunem, takže datetime.fromisoformat je převede přesně i v den změny času.
Python — REST
# Spotové ceny přes REST API (pip install requests)
from datetime import datetime, timedelta, timezone
import requests
API = "https://api.spotovky.cz"
def prices(day="today"):
"""Ceny dne: today, tomorrow, yesterday nebo YYYY-MM-DD. None, když data nejsou."""
r = requests.get(f"{API}/v2/prices/{day}", timeout=10)
if r.status_code == 404: # pro den zatím nejsou data (zítřek před ~13:05)
return None
r.raise_for_status() # 400 = neplatné datum
return r.json()
def current_slot(day):
now = datetime.now(timezone.utc)
for s in day["slots"]:
start = datetime.fromisoformat(s["start"]) # např. 2026-10-23T14:15:00+02:00
if start <= now < start + timedelta(minutes=15):
return s
return None
today = prices("today")
slot = current_slot(today) if today else None
if slot:
print(f"Aktuálně {slot['czk']:.2f} Kč/kWh, pořadí {slot['rank']}/{len(today['slots'])}")
tomorrow = prices("tomorrow")
print("Zítřek:", "zatím nezveřejněn" if tomorrow is None else f"{len(tomorrow['slots'])} intervalů")
Nejlevnější blok dne
Navazuje na předchozí ukázku. Hledá souvislý blok s nejnižší průměrnou cenou; počet intervalů dne bere z dat (92, 96 nebo 100).
Python — nejlevnější blok
def cheapest_block(day, hours):
"""Nejlevnější souvislý blok dané délky (např. pro pračku nebo nabíjení)."""
slots = day["slots"]
n = hours * 4 # 4 intervaly po 15 minutách = 1 hodina
if len(slots) < n:
return None
best = min(range(len(slots) - n + 1),
key=lambda i: sum(s["czk"] for s in slots[i:i + n]))
block = slots[best:best + n]
avg = sum(s["czk"] for s in block) / n
return block[0]["start"], avg
start, avg = cheapest_block(today, 2)
print(f"Nejlevnější 2 h od {start[11:16]}, průměr {avg:.2f} Kč/kWh")
MQTT s paho-mqtt
Broker spotovky.cz:1883 nevyžaduje jméno ani heslo. Ukázka odebírá aktuální interval ve formátu JSON i binárně; zprávy mají retain, takže přijdou hned po připojení a pak každých 15 minut.
Python — MQTT (paho-mqtt 2.x)
# Aktuální cena přes MQTT (pip install "paho-mqtt>=2")
import json
import struct
from datetime import datetime
from zoneinfo import ZoneInfo
import paho.mqtt.client as mqtt
TZ = ZoneInfo("Europe/Prague")
def on_connect(client, userdata, flags, reason_code, properties):
print("Připojeno:", reason_code)
client.subscribe([("v2/current/json", 1), ("v2/current/bin", 1)])
def on_message(client, userdata, msg):
if not msg.payload: # smazaná retain zpráva
return
if msg.topic == "v2/current/json":
d = json.loads(msg.payload)
print(f"JSON {d['start'][11:16]}–{d['end'][11:16]} {d['czk']:.2f} Kč/kWh pořadí {d['rank']}/{d['periods']}")
elif msg.topic == "v2/current/bin" and len(msg.payload) == 16 and msg.payload[0] == 2:
# verze, příznaky, začátek (UTC), haléře (int16), pořadí, kurz × 1000, počet intervalů
version, flags_, start, halere, rank, rate_x1000, periods = struct.unpack("<BBQhBHB", msg.payload)
t = datetime.fromtimestamp(start, TZ)
print(f"BIN {t:%H:%M} {halere / 100:.2f} Kč/kWh pořadí {rank}/{periods} kurz {rate_x1000 / 1000:.3f}")
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.on_connect = on_connect
client.on_message = on_message
client.connect("spotovky.cz", 1883, keepalive=60)
client.loop_forever()
Ukázka používá API knihovny paho-mqtt 2.x (CallbackAPIVersion.VERSION2). Broker přijímá jen subscribe na konkrétní topicy; publish a wildcardy jsou zakázané. Na Windows potřebuje zoneinfo ještě balíček tzdata (pip install tzdata).
Binární zpráva celého dne
Pro v2/today/bin a v2/tomorrow/bin. Počet intervalů N je na offsetu 12, čas intervalu je půlnoc + index × 900.
Python — dekódování dne
import struct
from datetime import datetime
from zoneinfo import ZoneInfo
def decode_day_v2(data: bytes):
"""v2/today/bin a v2/tomorrow/bin: 13 + N×3 bytů."""
version, flags, midnight, rate_x1000, n = struct.unpack_from("<BBQHB", data, 0)
if version != 2 or len(data) < 13 + n * 3:
raise ValueError("neočekávaný formát")
prices = struct.unpack_from(f"<{n}h", data, 13) # h = int16 se znaménkem
ranks = struct.unpack_from(f"<{n}B", data, 13 + n * 2)
tz = ZoneInfo("Europe/Prague")
for i, (halere, rank) in enumerate(zip(prices, ranks)):
if halere == -32768: # cena chybí
continue
yield datetime.fromtimestamp(midnight + i * 900, tz), halere / 100, rank