| |
| """Day-ahead price HISTORY for the EU zones from the ENTSO-E Transparency Platform (A44), cached as one parquet per |
| zone: hourly means by the zone's own wall clock (naive local timestamps, like the Nordic dataset's). |
| |
| Why ENTSO-E for all 25 zones, also those the site shows from Energy-Charts: one source and one definition of an hour for |
| the whole training set. Nothing is published from this data - the forecast is. |
| |
| The first run pulls history from START in 90-day requests (one zone, one request at a time, a pause between); later runs |
| re-pull only the last REFRESH_DAYS and merge, so a price that is revised is picked up. The token is read from the |
| environment (~/.config/elpriser.env) when a request is made, never logged and never part of a printed URL. |
| """ |
| import os |
| import sys |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
| from datetime import date, datetime, timedelta, timezone |
|
|
| import pandas as pd |
|
|
| DIR = os.path.dirname(os.path.abspath(__file__)) |
| sys.path.insert(0, DIR) |
| sys.path.insert(0, os.path.join(DIR, "..", "..", "eu_prices")) |
| import entsoe_parse as E |
| from zones import ZONES |
|
|
| SOURCE = "https://web-api.tp.entsoe.eu/api" |
| START = date(2024, 1, 1) |
| CHUNK_DAYS = 90 |
| REFRESH_DAYS = 10 |
| PAUSE = 1.5 |
| UA = "elpriser.org/1.0 (+https://elpriser.org; hello@elpriser.org)" |
|
|
|
|
| def token(): |
| t = os.environ.get("ENTSOE_TOKEN") |
| if not t: |
| raise RuntimeError("ENTSOE_TOKEN is not set - source ~/.config/elpriser.env") |
| return t |
|
|
|
|
| def fetch_a44(eic, start, end, tries=4): |
| """The A44 XML for [start, end) UTC midnights. Errors never carry the URL (it holds the token).""" |
| url = SOURCE + "?" + urllib.parse.urlencode({ |
| "securityToken": token(), "documentType": "A44", "in_Domain": eic, "out_Domain": eic, |
| "periodStart": start.strftime("%Y%m%d0000"), "periodEnd": end.strftime("%Y%m%d0000")}) |
| last = None |
| for attempt in range(tries): |
| try: |
| req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "application/xml"}) |
| with urllib.request.urlopen(req, timeout=120) as r: |
| return r.read().decode("utf-8", "replace") |
| except urllib.error.HTTPError as e: |
| last = f"HTTP {e.code}" |
| if e.code == 429: |
| time.sleep(min(int(e.headers.get("Retry-After", "30") or 30), 120) + 1) |
| elif e.code in (400, 401, 403): |
| raise RuntimeError(f"ENTSO-E {last}") |
| else: |
| time.sleep(10 * (attempt + 1)) |
| except (urllib.error.URLError, TimeoutError, OSError) as e: |
| last = type(e).__name__ |
| time.sleep(10 * (attempt + 1)) |
| raise RuntimeError(f"ENTSO-E failed after {tries} tries ({last})") |
|
|
|
|
| def hourly_local(slots, tzname): |
| """[(start_utc, minutes, price)] -> Series of hourly means on a naive LOCAL clock (duplicate autumn hour averaged, |
| the missing spring hour absent). 15-minute prices are averaged into their hour.""" |
| if not slots: |
| return pd.Series(dtype=float) |
| rows = [] |
| for start, minutes, price in slots: |
| for i in range(max(minutes // 15, 1)): |
| rows.append((start + timedelta(minutes=15 * i), price)) |
| df = pd.DataFrame(rows, columns=["t", "p"]) |
| df["t"] = pd.to_datetime(df.t, utc=True).dt.tz_convert(tzname).dt.tz_localize(None).dt.floor("h") |
| return df.groupby("t").p.mean().sort_index() |
|
|
|
|
| def pull_zone(zone, since, until): |
| eic, tzname, _ = ZONES[zone] |
| slots, cur = [], since |
| while cur < until: |
| nxt = min(cur + timedelta(days=CHUNK_DAYS), until) |
| slots += E.parse_a44(fetch_a44(eic, cur, nxt)) |
| cur = nxt |
| time.sleep(PAUSE) |
| return hourly_local(slots, tzname) |
|
|
|
|
| def update(zone, out_dir=DIR, today=None): |
| """Create or top up price_<zone>.parquet; returns the Series.""" |
| today = today or datetime.now(timezone.utc).date() |
| until = today + timedelta(days=2) |
| path = os.path.join(out_dir, f"price_{zone}.parquet") |
| if os.path.exists(path): |
| old = pd.read_parquet(path) |
| old = old.set_index(pd.to_datetime(old["t"]))["eur_mwh"] |
| since = max(old.index.max().date() - timedelta(days=REFRESH_DAYS), START) |
| new = pull_zone(zone, since, until) |
| merged = pd.concat([old[old.index < pd.Timestamp(since)], new]).sort_index() |
| merged = merged[~merged.index.duplicated(keep="last")] |
| else: |
| merged = pull_zone(zone, START, until) |
| merged.rename("eur_mwh").rename_axis("t").reset_index().to_parquet(path, index=False) |
| return merged |
|
|
|
|
| def main(argv=None): |
| zones = (argv or sys.argv[1:]) or list(ZONES) |
| for z in zones: |
| try: |
| s = update(z) |
| print(f"{z}: {len(s):,} hours, {s.index.min():%Y-%m-%d} -> {s.index.max():%Y-%m-%d}, " |
| f"mean {s.mean():.1f} EUR/MWh", flush=True) |
| except Exception as e: |
| print(f"{z}: FAILED {e}", flush=True) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
