# SPDX-License-Identifier: MIT """Build the figures and tables for YS-EN-0006 from results/*.npz and write them into the note between markers. Usage: python build_figures.py (after run_all.sh) Needs NumPy and pandas (for the hourly data in the example-week figures). """ from __future__ import annotations import datetime as dt import json import sys from pathlib import Path import numpy as np HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE.parent)) import svgchart as sc # noqa: E402 import analyse as A # noqa: E402 PAGE = HERE.parent.parent / "ys-en-0006-dilated-cnn-day-ahead-co2.html" RES = HERE / "results" RNG = np.random.default_rng(0) LABEL = { "naive_week": "Same hour last week", "naive_day": "Same hour yesterday", "naive_sametype": "Last day of the same type", "sametype_mean": "Mean of same-type days", "sametype_median": "Median of same-type days", "profile": "Room profile from training", "ridge": "Ridge regression", "ridge_cal": "Ridge + calendar", "ridge_cal_window": "Ridge + calendar, window norm", "linear_l1": "Linear, L1 loss", "linear_l1_cal": "Linear + calendar, L1 loss", "linear_l2_cal": "Linear + calendar, squared error", "tcn_report": "TCN, 2023 design", "tcn_report_cal": "TCN, 2023 design + calendar", "tcn": "TCN", "tcn_cal": "TCN + calendar", "tcn_cal_window": "TCN + calendar, window norm", "tcn_cal_pir": "TCN + calendar + motion sensor", "tcn_cal_l2": "TCN + calendar, squared-error loss", "gru_cal": "GRU + calendar", } KIND = {k: (1 if k.startswith(("tcn", "gru")) else 2 if k.startswith(("ridge", "linear")) else 0) for k in LABEL} def load(tag): p = RES / f"{tag}.npz" return A.load(p) if p.exists() else None def per_day(tag): truth, preds, lab, meta = load(tag) return {k: A.day_mae(v, truth)[0] for k, v in preds.items()}, lab, truth, preds, meta def boot_mean(v, dates): u, inv = np.unique(dates, return_inverse=True) s = np.bincount(inv, v, len(u)) n = np.bincount(inv, minlength=len(u)).astype(float) idx = RNG.integers(0, len(u), (2000, len(u))) m = s[idx].sum(1) / n[idx].sum(1) return float(v.mean()), float(np.percentile(m, 2.5)), float(np.percentile(m, 97.5)) # ---------------------------------------------------------------- figure: receptive field def fig_rf(): W, Hh, n, layers = 720, 300, 16, 4 x0, dx = 60, 31 ys = [262 - 52 * k for k in range(layers + 1)] o = [f''] # which nodes feed the final output used = [{n - 1}] for k in range(layers, 0, -1): d = 2 ** (k - 1) prev = set() for i in used[-1]: prev |= {i, i - d} used.append({i for i in prev if i >= 0}) used = used[::-1] # used[k] = nodes at level k on the final path for k in range(1, layers + 1): d = 2 ** (k - 1) for i in range(n): for j in (i, i - d): if j < 0: continue on = i in used[k] and j in used[k - 1] cls = "msg s1" if on else "rule" o.append(f'') for k in range(layers + 1): for i in range(n): f = "f1" if i in used[k] else "f0" r = 5 if i in used[k] else 3.5 o.append(f'') lab = "input (hours)" if k == 0 else f"layer {k} · dilation {2 ** (k - 1)}" o.append(f'{lab}') o.append(f't − 15') o.append(f't') o.append("") return sc.figure( code="FIG. 1 — HOW FAR A DILATED STACK SEES", fid="fig-rf", plot="\n".join(o), caption=("Figure 1. Causal convolutions with kernel size 2 and dilation doubling at each " "layer. The highlighted paths are every connection that reaches the last output: it " "depends on the last 24 = 16 inputs and on nothing later. With eight layers " "(dilations 1 to 128) the same construction sees 28 = 256 hours, enough for " "a week of hourly data."), ) # ---------------------------------------------------------------- figure: models on real data MAIN_ORDER = ["naive_week", "naive_day", "naive_sametype", "profile", "sametype_mean", "sametype_median", "ridge_cal", "linear_l1_cal", "tcn_report", "tcn_report_cal", "tcn", "tcn_cal", "gru_cal"] def vtt_all(): """Merge per-day MAE from every real-data run that shares the main test days.""" dm, lab, truth, preds, meta = per_day("vtt_main") extra = load("vtt_loss") if extra: t2, p2, l2, m2 = extra assert (l2["date"] == lab["date"]).all() for k, v in p2.items(): dm[k] = A.day_mae(v, t2)[0] preds[k] = v meta[k] = m2[k] return dm, lab, truth, preds, meta def hdots(rows, label, x_lo, x_hi, x_title, refs=(), ticks=None, fmt=lambda v: f"{v:.1f}"): """Horizontal dot + interval chart. rows = [(name, mean, lo, hi, slot)]; refs = [(x, text)] dashed reference lines.""" rh = 26 fr = sc.Frame(720, 40 + rh * len(rows) + 40, label, margins=(250, 40, 24, 40)) fr.x_scale(x_lo, x_hi) ticks = ticks or [t for t in range(int(x_lo), int(x_hi) + 1) if t % 5 == 0] fr.grid_x(ticks, fmt=lambda v: f"{v:g}") fr.titles(x_title) for i, (xv, txt) in enumerate(refs): x = fr.x(xv) fr.out.append(f'') fr.note(x + 4, fr.mt - 8 + 0 * i, txt) if i % 2 == 0 else fr.note(x - 4, fr.mt - 8, txt, anchor="end") for i, (name, m, lo, hi, slot) in enumerate(rows): y = fr.mt + rh * (i + 0.5) fr.out.append(f'{sc.esc(name)}') fr.out.append(f'') fr.out.append(f'') fr.out.append(f'{fmt(m)}') sc.point_hit(fr, fr.x(m), y, i, name, f"{slot}|MAE|{fmt(m)} ({fmt(lo)}–{fmt(hi)})") return fr.svg() def fig_models(): dm, lab, *_ = vtt_all() order = [k for k in MAIN_ORDER if k in dm] rows = [] for k in order: m, lo, hi = boot_mean(dm[k], lab["date"]) rows.append((LABEL[k], m, lo, hi, KIND[k])) best_base = min((r for r, k in zip(rows, order) if KIND[k] == 0), key=lambda r: r[1]) plot = hdots(rows, "Day-ahead forecast error on 780 test room-days for each method", 10, 30, "Mean absolute error, ppm CO2 (lower is better)", refs=[(best_base[1], f"best baseline {best_base[1]:.1f}")]) table = sc.table("Day-ahead MAE, VTT SCOTT test set (Oct–Dec 2019)", ["Method", "MAE, ppm", "95% interval", "Kind"], [[r[0], f"{r[1]:.2f}", f"{r[2]:.2f}–{r[3]:.2f}", ["baseline", "neural", "linear"][r[4] if r[4] < 3 else 0]] for r in rows]) return sc.figure( code="FIG. 4 — TOMORROW'S CO2, 13 ROOMS", fid="fig-models", plot=plot, legend_html=sc.legend([{"slot": 0, "name": "Baseline, no training"}, {"slot": 2, "name": "Linear model"}, {"slot": 1, "name": "Neural network"}]), caption=(f"Figure 4. Mean absolute error of each forecast over the 24 hours of the next " f"day, issued at midnight, on {len(lab['date'])} room-days from October to December 2019. " "Whiskers are 95% intervals from resampling whole dates. Neural rows average five " "training seeds. The vertical line marks the best baseline."), data_html=table, ) # ---------------------------------------------------------------- table: by day type def table_daytypes(): dm, lab, *_ = vtt_all() T = A.day_types(lab) xmas = lab["date"] >= "2019-12-23" cols = [("Workday", T["workday"] & ~xmas), ("First workday back", T["first workday back"] & ~xmas), ("Weekend", T["weekend"]), ("Weekday holiday", T["holiday on a weekday"]), ("23–31 Dec workday", T["workday"] & xmas), ("All", T["all"])] keys = [k for k in ["sametype_mean", "ridge_cal", "tcn_report", "tcn_report_cal", "tcn", "tcn_cal", "gru_cal"] if k in dm] head = ["Method"] + [f"{c} ({int(m.sum())})" for c, m in cols] rows = [[LABEL[k]] + [f"{dm[k][m].mean():.1f}" for _, m in cols] for k in keys] return sc.figure( code="FIG. 5 — WHERE THE GAIN IS", fid="fig-daytypes", plot=sc.table("Day-ahead MAE in ppm by kind of day", head, rows), caption=("Figure 5. Mean absolute error by kind of test day, room-days in brackets. The " "workday columns leave out 23–31 December, shown separately: those are working days on " "the national calendar on which the offices were empty. Neural rows average five seeds."), ) # ---------------------------------------------------------------- figure: example weeks def fig_week(fid, code, num, d0, d1, keys, room, caption): truth, preds, lab, meta = vtt_raw() keys_arr = load("vtt_main") # noqa: F841 (labels already loaded) z = np.load(RES / "vtt_main.npz") K = z["keys"] import exp_vtt as V rooms = sorted({s["name"] for s in V.series()}) ri = rooms.index(room) days = [d0 + dt.timedelta(k) for k in range((d1 - d0).days + 1)] xs = list(range(24 * len(days))) obs, fc = [None] * len(xs), {k: [None] * len(xs) for k in keys} for j, day in enumerate(days): t = 24 * (day - V.START).days hit = np.where((K[:, 0] == ri) & (K[:, 1] == t))[0] if len(hit): i = hit[0] for h in range(24): obs[24 * j + h] = float(truth[i, h]) for k in keys: fc[k][24 * j + h] = float(preds[k][:, i, h].mean()) fr = sc.Frame(720, 320, f"Observed CO2 and forecasts for {room}, {d0} to {d1}") vals = [v for v in obs if v is not None] + [v for k in keys for v in fc[k] if v is not None] lo, hi = 50 * (min(vals) // 50), 50 * (max(vals) // 50 + 1) fr.x_scale(0, len(xs) - 1) fr.y_scale(lo, hi) fr.grid_y(list(range(int(lo), int(hi) + 1, 50))) fr.x_labels([(24 * j + 12, d.strftime("%a %d")) for j, d in enumerate(days)]) for j in range(1, len(days)): fr.out.append(f'') fr.baseline() fr.titles("", "CO2, ppm") fr.cursor() series = [{"slot": 0, "name": "Observed", "values": obs}] for s, k in zip((1, 2, 3), keys): series.append({"slot": s, "name": LABEL[k], "values": fc[k], "end": LABEL[k].split(",")[0]}) for s in series[1:] + series[:1]: # forecasts first, observed on top; one path per day for j in range(len(days)): seg = s["values"][24 * j:24 * j + 24] if None in seg: continue d = "M" + " L".join(f"{fr.x(24 * j + h):.1f},{fr.y(v):.1f}" for h, v in enumerate(seg)) w = ' style="stroke-width:3"' if s["slot"] == 0 else "" fr.out.append(f'') sc.column_hits(fr, xs, series, lambda i, x: f"{days[i // 24].isoformat()} {i % 24:02d}:00", lambda v: f"{v:.0f} ppm", hover_marks=False) return sc.figure(code=code, fid=fid, plot=fr.svg(), caption=f"Figure {num}. " + caption, legend_html=sc.legend([{"slot": s["slot"], "name": s["name"]} for s in series])) _RAW = None def vtt_raw(): global _RAW if _RAW is None: truth, preds, lab, meta = load("vtt_main") _RAW = truth, preds, lab, meta return _RAW # ---------------------------------------------------------------- figure: synthetic SYN_ORDER = ["naive_week", "naive_day", "naive_sametype", "ridge", "ridge_cal", "tcn_report", "tcn", "tcn_report_cal", "tcn_cal", "gru_cal"] FLOOR_BEST, FLOOR_COPY = 0.2 * (2 / np.pi) ** 0.5, 0.2 * 2 ** 0.5 * (2 / np.pi) ** 0.5 def fig_synth(): dm, lab, *_ = per_day("synth_main") rows = [(LABEL[k], *boot_mean(dm[k], lab["date"]), KIND[k]) for k in SYN_ORDER if k in dm] plot = hdots(rows, "Day-ahead error on the synthetic office series for each method", 0, 1.2, "Mean absolute error (noise σ = 0.2; lower is better)", refs=[(FLOOR_COPY, "copy floor 0.226"), (FLOOR_BEST, "best possible 0.160")], ticks=[0, 0.2, 0.4, 0.6, 0.8, 1.0, 1.2], fmt=lambda v: f"{v:.3f}") table = sc.table("Day-ahead MAE, synthetic test half-year (two noise draws)", ["Method", "MAE", "95% interval", "Weekday holidays"], [[LABEL[k], f"{dm[k].mean():.3f}", f"{r[2]:.3f}–{r[3]:.3f}", f"{dm[k][(lab['holiday'] == 1) & (lab['dow'] < 5)].mean():.2f}"] for k, r in zip([k for k in SYN_ORDER if k in dm], rows)]) return sc.figure( code="FIG. 2 — THE CLEAN VERSION", fid="fig-synth", plot=plot, legend_html=sc.legend([{"slot": 0, "name": "Baseline, no training"}, {"slot": 2, "name": "Linear model"}, {"slot": 1, "name": "Neural network"}]), caption=("Figure 2. Mean absolute error on the rebuilt synthetic series, July–December " "2023, two independent noise draws. Dashed lines: the best any forecast can do (it knows " "the true profile) and the best a copy of one noisy past day can do (Equation 2). " "Neural rows average three seeds."), data_html=table, ) def fig_dow(): truth, preds, lab, meta = load("synth_hist") hol = lab["holiday"] == 1 ok = (~hol) & (lab["prev_workday"] == (((lab["dow"] - 1) % 7) < 5)) names = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] hs = [24, 72, 168] vals = {} for R in hs: e = np.abs(preds[f"tcn_h{R}"] - truth[None]).mean(-1).mean(0) vals[R] = [float(e[ok & (lab["dow"] == d)].mean()) for d in range(7)] ec = np.abs(preds["tcn_cal_h72"] - truth[None]).mean(-1).mean(0) cal = [float(ec[ok & (lab["dow"] == d)].mean()) for d in range(7)] fr = sc.Frame(720, 320, "TCN error on the synthetic series by day of the week, for three history lengths, without a calendar") fr.x_scale(-0.5, 6.5) fr.y_scale(0, 3) fr.grid_y([0, 0.5, 1, 1.5, 2, 2.5, 3], fmt=lambda v: f"{v:g}") fr.x_labels(list(enumerate(names))) fr.baseline() fr.titles("Day being forecast", "Mean absolute error") fr.cursor() series = [{"slot": 3, "name": "24 h of history", "values": vals[24], "end": "24 h"}, {"slot": 2, "name": "72 h", "values": vals[72], "end": "72 h"}, {"slot": 1, "name": "168 h", "values": vals[168], "end": "168 h"}, {"slot": 0, "name": "72 h + calendar", "values": cal, "end": "72 h + calendar"}] sc.draw_lines(fr, list(range(7)), series, end_labels=False) sc.column_hits(fr, list(range(7)), series, lambda i, x: names[i], lambda v: f"{v:.2f}") rows = [[f"{R} h"] + [f"{v:.2f}" for v in vals[R]] for R in hs] rows.append(["72 h + calendar"] + [f"{v:.2f}" for v in cal]) return sc.figure( code="FIG. 3 — COUNTING THE WEEK", fid="fig-dow", plot=fr.svg(), legend_html=sc.legend(series), caption=("Figure 3. Error of the TCN (no calendar) on the synthetic series by day of the " "week, when it sees only the last 24, 72 or 168 hours. Holidays and the days after them " "are left out. With 72 hours it cannot tell Friday from Saturday; given the calendar it " "doesn't need to. Three seeds each."), data_html=sc.table("MAE by day of the week", ["History"] + names, rows), ) # ---------------------------------------------------------------- figure: history, real data def fig_history(): dm, lab, *_ = per_day("vtt_hist") hs = [24, 48, 72, 120, 168] stats = [boot_mean(dm[f"tcn_cal_h{R}"], lab["date"]) for R in hs] fr = sc.Frame(720, 300, "TCN error on the real rooms against the length of history it reads") fr.x_scale(0, 180) fr.y_scale(14, 19) fr.grid_y([14, 15, 16, 17, 18, 19], fmt=lambda v: f"{v:g}") fr.grid_x(hs, fmt=lambda v: f"{v:g} h", lines=False) fr.baseline() fr.titles("Hours of history", "MAE, ppm") band = " L".join(f"{fr.x(R):.1f},{fr.y(s[2]):.1f}" for R, s in zip(hs, stats)) band += " L" + " L".join(f"{fr.x(R):.1f},{fr.y(s[1]):.1f}" for R, s in zip(reversed(hs), reversed(stats))) fr.out.append(f'') base = boot_mean(per_day("vtt_main")[0]["sametype_mean"], lab["date"])[0] fr.hline(base, cls="axis") fr.note(fr.x(4), fr.y(base) - 6, f"best baseline {base:.1f}") fr.cursor() series = [{"slot": 1, "name": "TCN + calendar", "values": [s[0] for s in stats]}] sc.draw_lines(fr, hs, series) sc.column_hits(fr, hs, series, lambda i, x: f"{x} h of history", lambda v: f"{v:.2f} ppm") return sc.figure( code="FIG. 8 — HOW MUCH HISTORY", fid="fig-history", plot=fr.svg(), caption=("Figure 8. Day-ahead error of the TCN with the calendar input when it reads only the " "last 24 to 168 hours. Band: 95% interval of the error level, resampling dates. Most of " "that width is day-to-day difficulty shared by every length; paired on the same days, 72 " "and 120 hours are within 0.3 ppm of 168 and 24 hours is 1.3 ppm worse (interval " "0.6–2.0). Five seeds each."), data_html=sc.table("MAE by history length", ["History", "MAE, ppm", "95% interval"], [[f"{R} h", f"{s[0]:.2f}", f"{s[1]:.2f}–{s[2]:.2f}"] for R, s in zip(hs, stats)]), ) # ---------------------------------------------------------------- figure: unseen levels def fig_level(): truth, preds, lab, meta = load("synth_trend") K = np.load(RES / "synth_trend.npz")["keys"] names = {"naive_sametype": "Last day of the same type", "ridge_cal_global": "Ridge + calendar", "ridge_cal_window": "Ridge + calendar, window norm", "tcn_cal_global": "TCN + calendar", "tcn_cal_window": "TCN + calendar, window norm"} rows = [] for k, nm in names.items(): e = np.abs(preds[k] - truth[None]).mean(-1).mean(0) rows.append([nm, f"{e[K[:, 0] == 0].mean():.2f}", f"{e[K[:, 0] == 1].mean():.2f}"]) return sc.figure( code="FIG. 9 — LEVELS IT HAS NEVER SEEN", fid="fig-level", plot=sc.table("MAE on synthetic series with a drifting level", ["Method", "Familiar level (0–10)", "Unseen level (15–25)"], rows), caption=("Figure 9. Synthetic series with a slowly wandering level, 0–10 in training. In " "the test half-year one series stays in that range and one moves to 15–25. Window norm " "subtracts each input week's mean and adds it back to the forecast. Three seeds."), ) # ---------------------------------------------------------------- table: verdict def fig_verdict(): rows = [ ["H1 · beats simple forecasts and a linear model", "Yes", "Direct head 15.5 vs 17.3 ppm (best baseline); the 2023 design 16.1; linear with the same loss 18.1"], ["H2 · a calendar fixes holidays at no cost", "Mostly, if the calendar is right", "Holidays 18.9 → 6.2 ppm; ordinary workdays +0.1; other first days back +1.0; " "23–31 Dec calendar workdays +5.9"], ["H3 · window norm fixes unseen levels", "Yes (synthetic)", "Unseen level 3.57 → 0.46; no effect on the real rooms"], ["H4 · a week beats 1–3 days", "Beats 1–2 days; 3 days ties", "With the calendar, vs 168 h: 24 h +1.3, 48 h +1.4, 72 h +0.3 (interval includes 0). " "Without one (synthetic only), Saturdays need 5 days"], ["H5 · beats a recurrent network", "No", "GRU 14.9 ppm, 0.6 better (interval 0.3–0.9); also closest to the synthetic noise floor"], ] return sc.figure( code="FIG. 10 — THE 2023 HYPOTHESES", fid="fig-verdict", plot=sc.table("What held up", ["Hypothesis", "Verdict", "Evidence"], rows, numeric_from=9), caption="Figure 10. The report's hypotheses against the evidence in this note.", ) def main(): figs = {"rf": fig_rf, "synth": fig_synth, "dow": fig_dow, "models": fig_models, "daytypes": table_daytypes, "history": fig_history, "level": fig_level, "verdict": fig_verdict} for name, fn in figs.items(): sc.inject(PAGE, name, fn()) sc.inject(PAGE, "holiday", fig_week( "fig-holiday", "FIG. 6 — INDEPENDENCE DAY", 6, dt.date(2019, 12, 2), dt.date(2019, 12, 8), ["sametype_mean", "tcn", "tcn_cal"], "room07", "One two-person office in the week of Finnish Independence Day, Friday 6 December. Each " "line segment is one midnight forecast of the next 24 hours; forecasts shown are a " "baseline and the TCN without and with the calendar input.")) sc.inject(PAGE, "xmas", fig_week( "fig-xmas", "FIG. 7 — BETWEEN THE YEARS", 7, dt.date(2019, 12, 23), dt.date(2019, 12, 31), ["sametype_mean", "tcn", "tcn_cal"], "room07", "The same office over Christmas. 24–26 December are public holidays; 23, 27, 30 and 31 " "December are working days on the national calendar, but the office stayed empty.")) if __name__ == "__main__": main()