File size: 3,656 Bytes
e49f88c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | #!/usr/bin/env python3
"""Great Britain's Elexon Market Index Price (MID) as an hourly history, cached in one parquet: price_gb.parquet (t = the hour on the London wall clock,
naive; eur_mwh = GBP/MWh - the column keeps the EU code's name, the unit is pounds).
The hours are exactly those the live job writes to KV `eu-prices-gb` (scripts/eu_prices/elexon_parse.py: the volume-weighted half-hour price, an
hour counts only when all its half hours are priced, hourly mean, London clock), so the model is trained on the number the page shows.
Elexon allows at most 7 days per request; this pulls 6 local days per request with a pause (a few hundred requests for four years, once). A later run
re-pulls only the last REFRESH_DAYS and merges. Licence: Contains BMRS data (c) Elexon Limited copyright and database right 2026 - the history is a
training input; nothing here is published except the model's own estimate.
python3 fetch_mid.py [--since 2023-01-01]
"""
import argparse
import os
import sys
import time
from datetime import date, datetime, timedelta, timezone
import pandas as pd
DIR = os.path.dirname(os.path.abspath(__file__))
for p in (DIR, os.path.join(DIR, "..", "..", "eu_prices"), os.path.join(DIR, "..", "eu")): # repo layout; flat on the Spark
sys.path.insert(0, p)
import elexon_parse as X # noqa: E402
import build_elexon_prices as BE # noqa: E402
TZ = "Europe/London"
START = date(2023, 1, 1)
CHUNK_DAYS = 6
REFRESH_DAYS = 10
PAUSE = 1.0
DATA = os.environ.get("GB_DATA_DIR", DIR) # where the caches live (the Spark: the job's own directory)
PATH = os.path.join(DATA, "price_gb.parquet")
def pull(since, until):
"""Hourly Series (naive London clock) for the local days since..until inclusive."""
rows, cur = {}, since
while cur <= until:
end = min(cur + timedelta(days=CHUNK_DAYS - 1), until)
dates = [(cur + timedelta(days=i)).isoformat() for i in range((end - cur).days + 1)]
a, b = BE.interval(dates, TZ)
slots = X.parse_mid(BE.fetch_mid(a, b))
for d in X.build_days(slots, TZ, dates, min_hours=1):
for p in d["prices"]:
if p["eur_mwh"] is not None:
rows[pd.Timestamp(d["date"]) + pd.Timedelta(hours=p["hour"])] = p["eur_mwh"]
cur = end + timedelta(days=1)
time.sleep(PAUSE)
return pd.Series(rows, dtype=float).sort_index()
def update(out=PATH, today=None, start=START, refresh=REFRESH_DAYS):
"""Create or top up the cache; returns the hourly Series."""
today = today or datetime.now(timezone.utc).date()
if os.path.exists(out):
old = pd.read_parquet(out)
old = old.set_index(pd.to_datetime(old["t"]))["eur_mwh"]
since = max(old.index.max().date() - timedelta(days=refresh), start)
new = pull(since, today)
merged = pd.concat([old[old.index < pd.Timestamp(since)], new])
else:
merged = pull(start, today)
merged = merged[~merged.index.duplicated(keep="last")].sort_index()
merged.rename("eur_mwh").rename_axis("t").reset_index().to_parquet(out, index=False)
return merged
def load(path=PATH):
df = pd.read_parquet(path)
s = df.set_index(pd.to_datetime(df["t"]))["eur_mwh"].sort_index()
return s[~s.index.duplicated(keep="last")]
if __name__ == "__main__":
ap = argparse.ArgumentParser()
ap.add_argument("--since", default=START.isoformat())
a = ap.parse_args()
s = update(start=date.fromisoformat(a.since))
print(f"gb: {len(s):,} hours, {s.index.min():%Y-%m-%d} -> {s.index.max():%Y-%m-%d}, mean {s.mean():.1f} GBP/MWh", flush=True)
|