| |
| """v3 training-set builder — adds spatial coverage and congestion signals. |
| |
| Same leakage discipline as v2 (see the TIME CONVENTION note there): one row |
| per (area, target hour, horizon 2-9); price lag k only populated when |
| k >= h-1; weather taken from the previous-runs archive at lead min(h, 7). |
| |
| What is new versus v2, and why: |
| |
| SPATIAL — v2 used one coordinate per area. Wind in Denmark is dominated by |
| offshore farms that sit well away from any inland point, and German solar |
| sits far south of German wind. v3 feeds the individual points as separate |
| features (plus a mean and a spread) rather than hand-picking weights: the |
| Phase 0 lesson on this project was that hand-picked constants are worse |
| than fitted ones, so the model is left to learn the weighting. |
| `wind_spread` (max-min across an area's points) is a genuine extra signal, |
| not just noise — a large spread means a weather front is crossing the area, |
| which is when production forecasts are least reliable. |
| |
| CONGESTION — two complementary signals, since JAO's own congestion figures |
| are unusable here (redistribution prohibited, and only ~1 day ahead): |
| 1. Neighbour-zone weather (NO2/SE3/SE4/NL) at the same forecast leads. |
| Congestion binds when a large price spread pushes flow past what the |
| border can carry, and those spreads are driven by neighbours' weather |
| — which IS available 10 days out. |
| 2. Month-ahead forecast transfer capacity (ENTSO-E A61/A03). This is the |
| capacity side of the same coin: scheduled maintenance and derating |
| that physically limit how much can flow, published a month out and so |
| legitimately available at every horizon here. (A first attempt used |
| A78 outage records instead; see fetch_ntc.py for why that was wrong.) |
| """ |
| import os |
| import numpy as np |
| import pandas as pd |
|
|
| P = os.path.dirname(os.path.abspath(__file__)) |
| BACKUP = os.path.expanduser("~/elpriser-data-backup") |
| HORIZONS = range(2, 10) |
|
|
| POINTS = { |
| "dk1_inland": (56.0, 9.5), "dk1_hornsrev": (55.5, 7.8), |
| "dk1_anholt": (56.6, 11.2), "dk1_south": (55.3, 9.2), |
| "dk2_zealand": (55.5, 12.0), "dk2_rodsand": (54.55, 11.7), "dk2_north": (56.0, 12.3), |
| "de_north": (54.0, 9.5), "de_central": (51.5, 10.0), "de_south": (48.5, 11.0), |
| "no2": (58.5, 7.0), "se3": (59.3, 17.0), "se4": (56.0, 14.0), "nl": (52.5, 5.0), |
| } |
|
|
| AREA_POINTS = { |
| "DK1": ["dk1_inland", "dk1_hornsrev", "dk1_anholt", "dk1_south"], |
| "DK2": ["dk2_zealand", "dk2_rodsand", "dk2_north"], |
| } |
| DE_POINTS = ["de_north", "de_central", "de_south"] |
| NEIGHBOURS = ["no2", "se3", "se4", "nl"] |
| AREA_BORDERS = { |
| "DK1": ["dk1_delu", "dk1_no2", "dk1_se3", "dk1_nl"], |
| "DK2": ["dk2_delu", "dk2_se4"], |
| } |
|
|
|
|
| def load_prices(area): |
| old = pd.read_parquet(f"{BACKUP}/eds_elspotprices_{area.lower()}.parquet") |
| old["t"] = pd.to_datetime(old["HourDK"]) |
| old = old[["t", "SpotPriceDKK"]].rename(columns={"SpotPriceDKK": "p"}) |
| new = pd.read_parquet(f"{BACKUP}/eds_dayaheadprices_{area.lower()}.parquet") |
| new["t"] = pd.to_datetime(new["TimeDK"]).dt.floor("h") |
| new = new.groupby("t", as_index=False)["DayAheadPriceDKK"].mean().rename( |
| columns={"DayAheadPriceDKK": "p"}) |
| df = pd.concat([old[old.t < "2025-10-01"], new[new.t >= "2025-10-01"]]) |
| return df.drop_duplicates("t").sort_values("t").set_index("t")["p"] |
|
|
|
|
| def load_production(area): |
| df = pd.read_parquet(f"{BACKUP}/eds_productionconsumptionsettlement_{area.lower()}.parquet") |
| df["t"] = pd.to_datetime(df["HourDK"]) |
| wind = ["OffshoreWindLt100MW_MWh", "OffshoreWindGe100MW_MWh", |
| "OnshoreWindLt50kW_MWh", "OnshoreWindGe50kW_MWh"] |
| sol = ["SolarPowerLt10kW_MWh", "SolarPowerGe10Lt40kW_MWh", |
| "SolarPowerGe40kW_MWh", "SolarPowerSelfConMWh"] |
| for c in wind + sol: |
| if c not in df.columns: |
| df[c] = 0.0 |
| df["prod"] = df[wind + sol].fillna(0).sum(axis=1) |
| return df.drop_duplicates("t").set_index("t")["prod"] |
|
|
|
|
| def w(kind, name): |
| df = pd.read_parquet(f"{P}/w_{kind}_{name}.parquet") |
| df["t"] = pd.to_datetime(df["time"]) |
| return df.set_index("t").drop(columns=["time"]) |
|
|
|
|
| def fit_estimator(weather_actual_points, target): |
| """Weather -> production, now on the multi-point mean (a better physical |
| proxy than a single site). Fitting on actuals is fine: the mapping is |
| static physics, and both training and inference rows feed it FORECAST |
| weather, so train/serve stay consistent.""" |
| import lightgbm as lgb |
| X = pd.concat([w("actual", n)[["wind_speed_100m", "direct_radiation"]] |
| for n in weather_actual_points]).groupby(level=0).mean() |
| df = X.join(target.rename("prod"), how="inner").dropna() |
| df["month"] = df.index.month |
| m = lgb.LGBMRegressor(n_estimators=150, num_leaves=31, learning_rate=0.08, verbosity=-1) |
| m.fit(df[["wind_speed_100m", "direct_radiation", "month"]], df["prod"]) |
| return m |
|
|
|
|
| def de_actual_production(): |
| dp = pd.read_parquet(f"{BACKUP}/entsoe_delu_generation_per_type.parquet") |
| dp = dp[dp.psr_type.isin(["B16", "B18", "B19"])] |
| dp["t"] = (pd.to_datetime(dp.datetime_utc, utc=True) |
| .dt.tz_convert("Europe/Copenhagen").dt.tz_localize(None).dt.floor("h")) |
| return dp.groupby(["t", "psr_type"]).quantity_mw.mean().groupby("t").sum() |
|
|
|
|
| def load_capacity(area, index): |
| """Month-ahead forecast transfer capacity per border and direction, scaled |
| by that border's own maximum so borders of very different sizes (Germany |
| ~1200 MW vs Sweden ~150 MW) are on one scale. 1.0 = full capacity. |
| |
| Month-ahead rather than week-ahead deliberately: it is published far |
| enough out to be legitimately available at h = 9, whereas week-ahead |
| would leak at the longest horizons. See fetch_ntc.py for why this |
| replaced an earlier attempt built on A78 outage records.""" |
| path = f"{P}/ntc_daily.parquet" |
| days = pd.Series(index.normalize(), index=index) |
| cols = {} |
| if not os.path.exists(path): |
| for b in AREA_BORDERS[area]: |
| for d in ("export", "import"): |
| cols[f"ntc_{b.split('_')[1]}_{d[:3]}"] = pd.Series(1.0, index=index) |
| return pd.DataFrame(cols, index=index) |
| n = pd.read_parquet(path) |
| n["date"] = pd.to_datetime(n["date"]) |
| for b in AREA_BORDERS[area]: |
| for d in ("export", "import"): |
| sub = n[(n.border == b) & (n.direction == d)] |
| key = f"ntc_{b.split('_')[1]}_{d[:3]}" |
| if not len(sub): |
| cols[key] = pd.Series(np.nan, index=index) |
| continue |
| byday = sub.set_index("date").mw |
| mx = byday.max() |
| scaled = (byday / mx) if mx > 0 else byday |
| cols[key] = days.map(scaled).astype(float).values |
| df = pd.DataFrame(cols, index=index) |
| exp = [c for c in df.columns if c.endswith("_exp")] |
| imp = [c for c in df.columns if c.endswith("_imp")] |
| df["ntc_exp_mean"] = df[exp].mean(axis=1) |
| df["ntc_imp_mean"] = df[imp].mean(axis=1) |
| df["ntc_min"] = df[exp + imp].min(axis=1) |
| return df |
|
|
|
|
| def build(area): |
| print(f"=== {area} ===", flush=True) |
| prices = load_prices(area) |
| prod = load_production(area) |
| pts = AREA_POINTS[area] |
|
|
| est = fit_estimator(pts, prod) |
| est_de = fit_estimator(DE_POINTS, de_actual_production()) |
| print(" estimators fitted", flush=True) |
|
|
| prev = {n: w("prev", n) for n in pts + DE_POINTS + NEIGHBOURS} |
| target_hours = prev[pts[0]].index |
| capacity = load_capacity(area, target_hours) |
|
|
| rows = [] |
| for h in HORIZONS: |
| lead = min(h, 7) |
| df = pd.DataFrame(index=target_hours) |
| df["y"] = prices.reindex(target_hours) |
| df["h"] = h |
| df["hour"] = df.index.hour |
| df["weekday"] = df.index.weekday |
| df["month"] = df.index.month |
| df["doy_sin"] = np.sin(2 * np.pi * df.index.dayofyear / 365.25) |
| df["doy_cos"] = np.cos(2 * np.pi * df.index.dayofyear / 365.25) |
| df["is_weekend"] = (df.weekday >= 5).astype(int) |
|
|
| for k in [1, 2, 3, 7, 14, 21, 28]: |
| col = prices.reindex(target_hours - pd.Timedelta(days=k)).values |
| df[f"lag{k}"] = col if k >= h - 1 else np.nan |
| last_known_day = df.index.normalize() - pd.Timedelta(days=1) * (h - 1) |
| day_mean = prices.groupby(prices.index.normalize()).mean() |
| df["last_day_mean"] = day_mean.reindex(last_known_day).values |
| df["seasonal4w"] = df[[f"lag{k}" for k in (7, 14, 21, 28)]].mean(axis=1) |
|
|
| |
| winds, rads = [], [] |
| for i, n in enumerate(pts): |
| pw = prev[n][f"wind_speed_100m_previous_day{lead}"].reindex(target_hours) |
| pr = prev[n][f"direct_radiation_previous_day{lead}"].reindex(target_hours) |
| pt_ = prev[n][f"temperature_2m_previous_day{lead}"].reindex(target_hours) |
| df[f"wind_p{i}"] = pw.values |
| df[f"rad_p{i}"] = pr.values |
| df[f"temp_p{i}"] = pt_.values |
| winds.append(pw.values); rads.append(pr.values) |
| for i in range(len(pts), 4): |
| df[f"wind_p{i}"] = np.nan; df[f"rad_p{i}"] = np.nan; df[f"temp_p{i}"] = np.nan |
| W = np.vstack(winds); R = np.vstack(rads) |
| df["wind"] = W.mean(axis=0) |
| df["rad"] = R.mean(axis=0) |
| df["wind_spread"] = W.max(axis=0) - W.min(axis=0) |
| df["temp"] = df[[f"temp_p{i}" for i in range(len(pts))]].mean(axis=1) |
|
|
| for i, n in enumerate(DE_POINTS): |
| df[f"wind_de{i}"] = prev[n][f"wind_speed_100m_previous_day{lead}"].reindex(target_hours).values |
| df[f"rad_de{i}"] = prev[n][f"direct_radiation_previous_day{lead}"].reindex(target_hours).values |
| df["wind_de"] = df[[f"wind_de{i}" for i in range(3)]].mean(axis=1) |
| df["rad_de"] = df[[f"rad_de{i}" for i in range(3)]].mean(axis=1) |
|
|
| for n in NEIGHBOURS: |
| df[f"wind_{n}"] = prev[n][f"wind_speed_100m_previous_day{lead}"].reindex(target_hours).values |
| df[f"rad_{n}"] = prev[n][f"direct_radiation_previous_day{lead}"].reindex(target_hours).values |
|
|
| df["est_prod"] = est.predict(pd.DataFrame({ |
| "wind_speed_100m": df.wind, "direct_radiation": df.rad, "month": df.month})) |
| df["est_prod_de"] = est_de.predict(pd.DataFrame({ |
| "wind_speed_100m": df.wind_de, "direct_radiation": df.rad_de, "month": df.month})) |
|
|
| for c in capacity.columns: |
| df[c] = capacity[c].values |
|
|
| rows.append(df.reset_index(names="t")) |
|
|
| out = pd.concat(rows, ignore_index=True).dropna(subset=["y"]) |
| out.to_parquet(f"{P}/dataset_v3_{area.lower()}.parquet", index=False) |
| print(f" saved dataset_v3_{area.lower()}.parquet: {len(out):,} rows, " |
| f"{len([c for c in out.columns if c not in ('t','y')])} features " |
| f"({out.t.min():%Y-%m-%d} -> {out.t.max():%Y-%m-%d})", flush=True) |
| return out |
|
|
|
|
| if __name__ == "__main__": |
| for a in ["DK1", "DK2"]: |
| build(a) |
| print("DATASET DONE", flush=True) |
|
|
