File size: 5,249 Bytes
5d7f266 | 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 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | #!/usr/bin/env python3
"""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")) # in the repo; flat next to this file on Spark
import entsoe_parse as E # noqa: E402
from zones import ZONES # noqa: E402
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) # tomorrow's auction is published ~13:00 CET
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: # one zone must not stop the others
print(f"{z}: FAILED {e}", flush=True)
if __name__ == "__main__":
main()
|