Log forecast snapshots and scoreboard Toto vs NWS

- src/forecast_log.py: SQLite tables for forecast_snapshots and actuals,
upsert helpers, and a scoreboard query computing rolling MAE per source
using each target's most-recently-issued forecast.
- app.py: write actuals + Toto + NWS forecasts on every refresh; render a
markdown scoreboard panel above the plots; daemon thread that re-runs
refresh() hourly so snapshots accumulate without a user visit.
- app.py: drop the NWS first period that overlaps Ecowitt's last actual
hour so all forecasts begin one bucket past the last observed value.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app.py

@@ -9,19 +9,22 @@ forecast for the same window.
 from __future__ import annotations
 
 import os
+import threading
 import time
+import traceback
 from datetime import datetime, timedelta, timezone
 
 import gradio as gr
 import pandas as pd
 
-from src import ecowitt, nws
+from src import ecowitt, forecast_log, nws
 from src.forecast import forecast_series
 from src.plotting import metric_figure
 
 CACHE_TTL_SECONDS = 60 * 60  # 1 hour
 HISTORY_DAYS = 7
 HORIZON_HOURS = 24
+AUTO_REFRESH_SECONDS = 60 * 60  # log a fresh forecast snapshot every hour
 
 # Three metrics to forecast. Maps Ecowitt history column → plot config.
 METRICS = [
@@ -73,13 +76,26 @@ def refresh():
     nws_df = fetch_nws()
     now = pd.Timestamp.now(tz="UTC").floor("h")
 
+    # NWS's first period and Ecowitt's last bucket describe the same wall-clock
+    # hour; drop the overlap so all forecasts begin one hour after the last
+    # observed actual.
+    last_actual = history.dropna(how="all").index.max()
+    nws_future = nws_df[nws_df.index > last_actual] if last_actual is not None else nws_df
+
+    log_conn = forecast_log.connect()
+    forecast_log.record_actuals(log_conn, history)
+
     figs = []
     for m in METRICS:
         series = history[m["col"]].dropna()
         toto = forecast_series(series, horizon=HORIZON_HOURS)
-        nws_series = (
-            nws_df[m["nws_col"]] if (m["nws_col"] and m["nws_col"] in nws_df.columns) else None
-        )
+        forecast_log.record_toto(log_conn, m["col"], toto)
+
+        nws_series = None
+        if m["nws_col"] and m["nws_col"] in nws_future.columns:
+            nws_series = nws_future[m["nws_col"]].dropna()
+            forecast_log.record_nws(log_conn, m["col"], nws_series)
+
         figs.append(
             metric_figure(
                 history=series.tail(HISTORY_DAYS * 24),
@@ -90,7 +106,59 @@ def refresh():
                 now=now,
             )
         )
-    return figs[0], figs[1], figs[2]
+    scoreboard_md = render_scoreboard(log_conn)
+    return figs[0], figs[1], figs[2], scoreboard_md
+
+
+# --- scoreboard ----------------------------------------------------------
+def render_scoreboard(conn) -> str:
+    lines = ["### Forecast scoreboard (rolling 48h MAE, lower = better)"]
+    any_data = False
+    for metric, label, unit in [
+        ("temp_f", "Temperature", "°F"),
+        ("humidity", "Humidity", "%"),
+        ("pressure_inhg", "Pressure", "inHg"),
+    ]:
+        summ = forecast_log.scoreboard_summary(conn, metric=metric, window_hours=48)
+        if summ.empty:
+            continue
+        any_data = True
+        by = {row["source"]: row for _, row in summ.iterrows()}
+        toto = by.get("toto")
+        nws = by.get("nws")
+        parts = [f"**{label}**"]
+        if toto is not None:
+            parts.append(f"Toto {toto['mae']:.2f} {unit} (n={int(toto['n'])})")
+        if nws is not None:
+            parts.append(f"NWS {nws['mae']:.2f} {unit} (n={int(nws['n'])})")
+        if toto is not None and nws is not None:
+            diff = toto["mae"] - nws["mae"]
+            winner = "Toto" if diff < 0 else "NWS"
+            parts.append(f"→ **{winner}** by {abs(diff):.2f} {unit}")
+        lines.append(" · ".join(parts))
+    if not any_data:
+        lines.append("_No scored forecasts yet — the scoreboard fills in once forecasts have target hours in the past with matching Ecowitt actuals (typically after the first hour of running)._")
+    return "\n\n".join(lines)
+
+
+# --- auto-refresh background thread --------------------------------------
+def _autorefresh_loop():
+    """Call refresh() on a schedule so we accumulate forecast snapshots even
+    when nobody is loading the page. Errors are logged and swallowed so a
+    transient API failure doesn't kill the thread."""
+    while True:
+        try:
+            refresh()
+        except Exception:  # noqa: BLE001
+            print("[autorefresh] error during refresh:")
+            traceback.print_exc()
+        time.sleep(AUTO_REFRESH_SECONDS)
+
+
+def _start_autorefresh():
+    t = threading.Thread(target=_autorefresh_loop, daemon=True, name="autorefresh")
+    t.start()
+    print(f"[autorefresh] started, interval={AUTO_REFRESH_SECONDS}s")
 
 
 # --- UI -------------------------------------------------------------------
@@ -109,14 +177,16 @@ with gr.Blocks(title="Toto Weather Forecast") as demo:
     gr.Markdown(HOOK)
     gr.Markdown(SUBTITLE)
     refresh_btn = gr.Button("Refresh forecast", variant="primary")
+    scoreboard_md = gr.Markdown()
     temp_plot = gr.Plot(label="Temperature")
     humidity_plot = gr.Plot(label="Humidity")
     pressure_plot = gr.Plot(label="Pressure")
 
-    outputs = [temp_plot, humidity_plot, pressure_plot]
+    outputs = [temp_plot, humidity_plot, pressure_plot, scoreboard_md]
     demo.load(refresh, outputs=outputs)
     refresh_btn.click(refresh, outputs=outputs)
 
 
 if __name__ == "__main__":
+    _start_autorefresh()
     demo.launch()

src/forecast_log.py

@@ -0,0 +1,175 @@
+"""SQLite logging of Toto + NWS forecasts and Ecowitt actuals.
+
+Every refresh appends:
+  - actuals(target_ts, metric, value)            ← Ecowitt history rows
+  - forecast_snapshots(forecast_made_at, target_ts, source, metric, p10, p50, p90)
+                                                  ← one row per future hour, per source, per metric
+
+A scoreboard joins the two and computes per-source MAE over a rolling window.
+
+NOTE: On HuggingFace Spaces' free CPU tier the DB lives in ephemeral storage
+and resets when the Space rebuilds (i.e. on `git push`). Restarts without
+rebuild keep the file. For longer-lived tracking, mount persistent storage
+or push to an HF Dataset.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import time
+from typing import Iterable
+
+import pandas as pd
+
+from .forecast import TotoForecast
+
+DEFAULT_DB_PATH = os.environ.get("FORECAST_LOG_DB", "data/forecasts.db")
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS forecast_snapshots (
+    forecast_made_at INTEGER NOT NULL,
+    target_ts        INTEGER NOT NULL,
+    source           TEXT    NOT NULL,        -- 'toto' | 'nws'
+    metric           TEXT    NOT NULL,        -- 'temp_f' | 'humidity' | 'pressure_inhg'
+    p10              REAL,
+    p50              REAL,
+    p90              REAL,
+    PRIMARY KEY (forecast_made_at, target_ts, source, metric)
+);
+CREATE INDEX IF NOT EXISTS idx_fs_target ON forecast_snapshots(target_ts, metric, source);
+
+CREATE TABLE IF NOT EXISTS actuals (
+    target_ts INTEGER NOT NULL,
+    metric    TEXT    NOT NULL,
+    value     REAL    NOT NULL,
+    PRIMARY KEY (target_ts, metric)
+);
+"""
+
+
+def connect(path: str = DEFAULT_DB_PATH) -> sqlite3.Connection:
+    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
+    conn = sqlite3.connect(path)
+    conn.executescript(SCHEMA)
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA synchronous=NORMAL")
+    return conn
+
+
+def _ts(t) -> int:
+    return int(pd.Timestamp(t).tz_convert("UTC").timestamp())
+
+
+def record_actuals(conn: sqlite3.Connection, history: pd.DataFrame) -> int:
+    """Upsert actuals from a history DataFrame (UTC-indexed; column = metric)."""
+    rows = []
+    for metric in history.columns:
+        s = history[metric].dropna()
+        for ts, val in s.items():
+            rows.append((_ts(ts), metric, float(val)))
+    if not rows:
+        return 0
+    conn.executemany(
+        "INSERT OR REPLACE INTO actuals (target_ts, metric, value) VALUES (?,?,?)",
+        rows,
+    )
+    conn.commit()
+    return len(rows)
+
+
+def record_toto(
+    conn: sqlite3.Connection,
+    metric: str,
+    fcst: TotoForecast,
+    forecast_made_at: int | None = None,
+) -> int:
+    made = forecast_made_at if forecast_made_at is not None else int(time.time())
+    rows = [
+        (made, _ts(t), "toto", metric, float(p10), float(p50), float(p90))
+        for t, p10, p50, p90 in zip(fcst.median.index, fcst.p10.values, fcst.median.values, fcst.p90.values)
+    ]
+    conn.executemany(
+        "INSERT OR REPLACE INTO forecast_snapshots "
+        "(forecast_made_at, target_ts, source, metric, p10, p50, p90) "
+        "VALUES (?,?,?,?,?,?,?)",
+        rows,
+    )
+    conn.commit()
+    return len(rows)
+
+
+def record_nws(
+    conn: sqlite3.Connection,
+    metric: str,
+    series: pd.Series,
+    forecast_made_at: int | None = None,
+) -> int:
+    """NWS gives a point forecast only — store as p50 with NULL p10/p90."""
+    made = forecast_made_at if forecast_made_at is not None else int(time.time())
+    s = series.dropna()
+    rows = [(made, _ts(t), "nws", metric, None, float(v), None) for t, v in s.items()]
+    if not rows:
+        return 0
+    conn.executemany(
+        "INSERT OR REPLACE INTO forecast_snapshots "
+        "(forecast_made_at, target_ts, source, metric, p10, p50, p90) "
+        "VALUES (?,?,?,?,?,?,?)",
+        rows,
+    )
+    conn.commit()
+    return len(rows)
+
+
+def scoreboard(
+    conn: sqlite3.Connection,
+    metric: str = "temp_f",
+    window_hours: int = 48,
+) -> pd.DataFrame:
+    """Per-source MAE over the last `window_hours`, restricted to forecasts
+    whose target time is in the past and where we have an actual value.
+
+    Each (target_ts, source) pair is scored against the *most recent* forecast
+    issued for that target — i.e. the latest snapshot before target_ts.
+    """
+    cutoff = int(time.time()) - window_hours * 3600
+    sql = """
+    WITH latest AS (
+        SELECT source, target_ts, metric,
+               MAX(forecast_made_at) AS forecast_made_at
+        FROM forecast_snapshots
+        WHERE metric = ?
+          AND forecast_made_at <= target_ts
+          AND target_ts <= ?
+          AND target_ts >= ?
+        GROUP BY source, target_ts, metric
+    )
+    SELECT f.source,
+           f.target_ts,
+           f.p50         AS prediction,
+           a.value       AS actual,
+           ABS(f.p50 - a.value) AS abs_err
+    FROM forecast_snapshots f
+    JOIN latest l USING (source, target_ts, metric, forecast_made_at)
+    JOIN actuals a USING (target_ts, metric)
+    """
+    now = int(time.time())
+    df = pd.read_sql_query(sql, conn, params=[metric, now, cutoff])
+    if df.empty:
+        return df
+    return df
+
+
+def scoreboard_summary(
+    conn: sqlite3.Connection,
+    metric: str = "temp_f",
+    window_hours: int = 48,
+) -> pd.DataFrame:
+    df = scoreboard(conn, metric=metric, window_hours=window_hours)
+    if df.empty:
+        return pd.DataFrame(columns=["source", "n", "mae"])
+    return (
+        df.groupby("source")["abs_err"]
+        .agg(n="count", mae="mean")
+        .reset_index()
+    )