# 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'")
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()