| """Models of the Great Britain estimate on the rows of gbfeat.py. Pure functions (numpy / pandas, LightGBM imported where used). |
| |
| A level + shape b_A (the naive method every candidate must beat; also the EU model's) |
| B intraday update b_A + a ridge on the realised residuals of today's hours so far (decay-weighted), per (as-of hour, day offset); |
| b_simple / b_mult two fixed rules with no fitted number but the decay: add the mean residual of the last three hours, |
| resp. scale the shape by the ratio of realised to expected so far, both fading with the hours ahead |
| C ridge + LightGBM on the residual y - b_A from the full feature set (weather, calendar, lags, intraday); the EU recipe's two models (a rolling |
| 365-day ridge and LightGBM, mean of the two), retrained on Great Britain with the intraday features added |
| band LightGBM quantile models (P10 / P90 of y - b_A) widened by 1.2 x the rolling 10th / 90th percentile of their own miss |
| (the EU quantile-edge recipe, scripts/forecast_model/eu/calibrate_bands.py) |
| """ |
| import os |
| import sys |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| DIR = os.path.dirname(os.path.abspath(__file__)) |
| sys.path.insert(0, DIR) |
| import gbfeat as GF |
|
|
| LGB_PARAMS = dict(objective="quantile", alpha=0.5, n_estimators=300, num_leaves=31, min_child_samples=100, learning_rate=0.06, subsample=0.6, |
| subsample_freq=1, colsample_bytree=0.8, max_bin=127, verbosity=-1) |
| RIDGE_ALPHA_REL = 0.005 |
| RIDGE_WINDOW_DAYS = 365 |
| MIN_TRAIN_ROWS = 400 |
| BAND_SCALE = 1.2 |
| BAND_WINDOW_DAYS = 180 |
|
|
|
|
| |
| def b_simple(df, tau=6.0): |
| return df["b_A"] + df["rho3"] * np.exp(-df["ahead"] / tau) |
|
|
|
|
| def b_mult(df, tau=12.0): |
| r = (df["known_mean"] / df["a_known_mean"]).where(df["a_known_mean"] > 20).clip(0.5, 2.0).fillna(1.0) |
| return df["b_A"] * (1 + (r - 1) * np.exp(-df["ahead"] / tau)) |
|
|
|
|
| |
| def ridge_fit(X, y, alpha_rel=RIDGE_ALPHA_REL): |
| mu = X.mean(axis=0) |
| sd = X.std(axis=0) |
| sd[sd < 1e-9] = 1.0 |
| Z = (X - mu) / sd |
| ym = y.mean() |
| A = Z.T @ Z + alpha_rel * len(y) * np.eye(X.shape[1]) |
| return mu, sd, np.linalg.solve(A, Z.T @ (y - ym)), ym |
|
|
|
|
| def ridge_predict(m, X): |
| mu, sd, beta, b0 = m |
| return b0 + ((X - mu) / sd) @ beta |
|
|
|
|
| def fit_ridges(train, design, window_days=RIDGE_WINDOW_DAYS, extra_mask=None): |
| """{(a, k): model}: one ridge per as-of hour and day offset on the last `window_days` of settled targets (the residual y - b_A).""" |
| models = {} |
| end = train["date"].max() |
| tr = train[train["date"] > end - pd.Timedelta(days=window_days)] |
| for (a, k), g in tr.groupby(["a", "k"]): |
| g = g.dropna(subset=["y", "b_A"]) |
| if len(g) < MIN_TRAIN_ROWS: |
| continue |
| X = design(g) |
| Xv = X.to_numpy(float) |
| ok = np.isfinite(Xv).all(axis=1) |
| if ok.sum() < MIN_TRAIN_ROWS: |
| continue |
| models[(int(a), int(k))] = (list(X.columns), ridge_fit(Xv[ok], (g["y"] - g["b_A"]).to_numpy(float)[ok])) |
| return models |
|
|
|
|
| def predict_ridges(models, df, design): |
| """b_A + the ridge's correction for every row of df whose (a, k) has a model; NaN elsewhere. The as-of hour of a row without its own model uses the |
| nearest trained one at or below it (a job that runs at every hour, trained at a few).""" |
| out = pd.Series(np.nan, index=df.index) |
| trained_a = sorted({a for a, _ in models}) |
| for (a, k), g in df.groupby(["a", "k"]): |
| use = max([t for t in trained_a if t <= a], default=None) |
| if use is None or (use, int(k)) not in models: |
| continue |
| cols, m = models[(use, int(k))] |
| X = design(g)[cols].to_numpy(float) |
| ok = np.isfinite(X).all(axis=1) & g["b_A"].notna().to_numpy() |
| if ok.any(): |
| out.loc[g.index[ok]] = g["b_A"].to_numpy()[ok] + ridge_predict(m, X[ok]) |
| return out |
|
|
|
|
| |
| def fit_lgb(train, alpha=0.5, jobs=8, feats=GF.LGB_FEATS, params=None): |
| import lightgbm as lgb |
| tr = train.dropna(subset=["y", "b_A"]) |
| m = lgb.LGBMRegressor(n_jobs=jobs, **{**LGB_PARAMS, "alpha": alpha, **(params or {})}) |
| m.fit(tr[feats], tr["y"] - tr["b_A"]) |
| return m |
|
|
|
|
| def predict_lgb(m, df, feats=GF.LGB_FEATS): |
| return pd.Series(m.predict(df[feats]) + df["b_A"].to_numpy(), index=df.index) |
|
|
|
|
| |
| def lead_class(df): |
| """The class a band offset belongs to: the rest of today by hours ahead (1-3, 4-8, 9-24), then the day offset k = 1 .. 7.""" |
| k = df["k"].to_numpy() |
| ah = df["ahead"].to_numpy() |
| return np.where(k == 0, np.where(ah <= 3, "r1", np.where(ah <= 8, "r2", "r3")), np.char.add("d", k.astype(str))) |
|
|
|
|
| def band_offsets(res, window_days=BAND_WINDOW_DAYS, scale=BAND_SCALE, min_n=200): |
| """{class: {lo, hi}} from a frame with `cls`, `date` (target day), `y`, `q10`, `q90`: scale x the 10th percentile of y - q10 and the 90th of y - q90 |
| over the last `window_days` of target days in the frame (all settled).""" |
| d = res.dropna(subset=["y", "q10", "q90"]) |
| if d.empty: |
| return {} |
| d = d[d["date"] > d["date"].max() - pd.Timedelta(days=window_days)] |
| out = {} |
| for c, g in d.groupby("cls"): |
| if len(g) < min_n: |
| continue |
| out[str(c)] = {"lo": round(float(np.percentile(g["y"] - g["q10"], 10)) * scale, 2), "hi": round(float(np.percentile(g["y"] - g["q90"], 90)) * scale, 2)} |
| return out |
|
|
|
|
| def apply_band(df, point, q10, q90, offsets): |
| """Absolute P10 / P90 arrays: the quantile models' predictions moved by their class offsets and widened to contain the point estimate.""" |
| cls = lead_class(df) |
| lo = np.array([offsets.get(c, {}).get("lo", np.nan) for c in cls]) |
| hi = np.array([offsets.get(c, {}).get("hi", np.nan) for c in cls]) |
| p = np.asarray(point, float) |
| return np.minimum(np.asarray(q10, float) + lo, p), np.maximum(np.asarray(q90, float) + hi, p) |
|
|
