feat: forecast precompute + issued-forecast archive + dashboard overlay

The collection-leader worker now precomputes forecasts after every
scrape cycle: primes the /forecast cache (users never trigger the
multi-second inference — its TTL rises to 4500s so the hourly refresh
always wins) and persists every issued forecast to a new
forecast_history table keyed by (as_of, station, horizon) with
predicted max level, warn/danger probabilities, current level, and
model_version. This is the operational record the backtests lacked —
predicted-vs-actual becomes a simple join instead of retraining
historical models.

GET /api/forecast/history/{station} serves the archive (hours or
start/end + horizon filters, 5-min edge cache), and the station history
chart overlays 'Model 24 h peak (as issued)' as a dashed violet line
once data accumulates.

src/forecast_history.py

@@ -0,0 +1,172 @@
+"""Persistence for issued flood forecasts.
+
+Every background precompute stores what the deployed model predicted at that
+moment — predicted 24/12/6 h peak, warning/danger probabilities, model
+version. Keyed by (as_of, station, horizon), so hourly data yields one row
+per station-horizon per hour regardless of how often the precompute runs.
+This is the operational record that lets "predicted vs actual" be graphed
+later without retraining historical models.
+"""
+
+import datetime
+import logging
+from typing import Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+class ForecastHistoryStore:
+    """SQL store (sqlite / postgresql / mysql), same pattern as HiiStore."""
+
+    def __init__(self, connection_string: str, db_type: str):
+        self.db_type = db_type.lower()
+        if self.db_type not in ("sqlite", "postgresql", "mysql"):
+            raise ValueError(
+                f"Forecast history requires a SQL database, got '{db_type}'"
+            )
+        self.connection_string = connection_string
+        self.engine = None
+
+    def connect(self) -> bool:
+        try:
+            from sqlalchemy import create_engine, text
+
+            self.engine = create_engine(self.connection_string, pool_pre_ping=True)
+            ddl = """
+                CREATE TABLE IF NOT EXISTS forecast_history (
+                    as_of TIMESTAMP NOT NULL,
+                    station_code VARCHAR(10) NOT NULL,
+                    horizon_hours INTEGER NOT NULL,
+                    predicted_max_level NUMERIC(8,3),
+                    p_warning NUMERIC(7,5),
+                    p_danger NUMERIC(7,5),
+                    current_level NUMERIC(8,3),
+                    model_version VARCHAR(64),
+                    source VARCHAR(16),
+                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+                    PRIMARY KEY (as_of, station_code, horizon_hours)
+                )
+            """
+            index = (
+                "CREATE INDEX IF NOT EXISTS idx_forecast_history_station "
+                "ON forecast_history(station_code, as_of)"
+            )
+            with self.engine.begin() as conn:
+                conn.execute(text(ddl))
+                if self.db_type != "mysql":  # MySQL lacks IF NOT EXISTS for indexes
+                    conn.execute(text(index))
+            return True
+        except Exception as error:
+            logger.error(f"ForecastHistoryStore failed to connect: {error}")
+            self.engine = None
+            return False
+
+    def save_rows(self, rows: List[Dict]) -> int:
+        """Upsert forecast rows as returned by ml.predict (idempotent)."""
+        if not rows:
+            return 0
+        if not self.engine and not self.connect():
+            return 0
+        from sqlalchemy import text
+
+        cols = (
+            "(as_of, station_code, horizon_hours, predicted_max_level, "
+            "p_warning, p_danger, current_level, model_version, source)"
+        )
+        values = (
+            "(:as_of, :station_code, :horizon_hours, :predicted_max_level, "
+            ":p_warning, :p_danger, :current_level, :model_version, :source)"
+        )
+        update_cols = (
+            "predicted_max_level",
+            "p_warning",
+            "p_danger",
+            "current_level",
+            "model_version",
+            "source",
+        )
+        if self.db_type == "mysql":
+            updates = ", ".join(f"{c} = VALUES({c})" for c in update_cols)
+            sql = (
+                f"INSERT INTO forecast_history {cols} VALUES {values} "
+                f"ON DUPLICATE KEY UPDATE {updates}"
+            )
+        else:
+            updates = ", ".join(f"{c} = EXCLUDED.{c}" for c in update_cols)
+            sql = (
+                f"INSERT INTO forecast_history {cols} VALUES {values} "
+                f"ON CONFLICT (as_of, station_code, horizon_hours) "
+                f"DO UPDATE SET {updates}"
+            )
+
+        params = []
+        for row in rows:
+            as_of = row.get("as_of")
+            if isinstance(as_of, str):
+                as_of = datetime.datetime.fromisoformat(as_of)
+            if as_of is None or row.get("station_code") is None:
+                continue
+            params.append(
+                {
+                    "as_of": as_of,
+                    "station_code": row["station_code"],
+                    "horizon_hours": row.get("horizon_hours"),
+                    "predicted_max_level": row.get("predicted_max_level"),
+                    "p_warning": row.get("p_warning"),
+                    "p_danger": row.get("p_danger"),
+                    "current_level": row.get("current_level"),
+                    "model_version": row.get("model_version"),
+                    "source": row.get("source"),
+                }
+            )
+        if not params:
+            return 0
+        try:
+            with self.engine.begin() as conn:
+                conn.execute(text(sql), params)
+            return len(params)
+        except Exception as error:
+            logger.error(f"ForecastHistoryStore save failed: {error}")
+            return 0
+
+    def fetch(
+        self,
+        station_code: str,
+        start: Optional[datetime.datetime] = None,
+        end: Optional[datetime.datetime] = None,
+        horizon_hours: Optional[int] = None,
+        limit: int = 5000,
+    ) -> List[Dict]:
+        """Issued forecasts for one station, ascending by as_of."""
+        if not self.engine and not self.connect():
+            return []
+        from sqlalchemy import text
+
+        clauses = ["station_code = :code"]
+        params: Dict = {"code": station_code, "limit": limit}
+        if start is not None:
+            clauses.append("as_of >= :start")
+            params["start"] = start
+        if end is not None:
+            clauses.append("as_of <= :end")
+            params["end"] = end
+        if horizon_hours is not None:
+            clauses.append("horizon_hours = :horizon")
+            params["horizon"] = horizon_hours
+        sql = (
+            "SELECT as_of, station_code, horizon_hours, predicted_max_level, "
+            "p_warning, p_danger, current_level, model_version, source "
+            f"FROM forecast_history WHERE {' AND '.join(clauses)} "
+            "ORDER BY as_of ASC, horizon_hours ASC LIMIT :limit"
+        )
+        try:
+            with self.engine.connect() as conn:
+                rows = [dict(r._mapping) for r in conn.execute(text(sql), params)]
+            for row in rows:
+                for key, value in row.items():
+                    if hasattr(value, "is_finite"):  # Decimal -> float
+                        row[key] = float(value)
+            return rows
+        except Exception as error:
+            logger.error(f"ForecastHistoryStore fetch failed: {error}")
+            return []

src/static/dashboard.html

@@ -633,6 +633,34 @@
             };
             const sampled = rows.length > 2000 ? downsample(rows) : rows;
 
+            // Issued model predictions (recorded hourly by the server) —
+            // overlays what the model said would happen, when it said it.
+            let predSeries = null;
+            try {
+                const fr = await fetch(`/api/forecast/history/${encodeURIComponent(stationCode)}?${query}&horizon=24`);
+                const forecastRows = fr.ok ? await fr.json() : [];
+                if (forecastRows.length) {
+                    const keyOf = (value) => {
+                        const d = new Date(value);
+                        return rows.length > 2000
+                            ? `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}`
+                            : `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}-${d.getUTCHours()}`;
+                    };
+                    const byKey = new Map();
+                    forecastRows.forEach((r) => {
+                        if (r.predicted_max_level == null) return;
+                        const key = keyOf(r.as_of);
+                        if (!byKey.has(key)) byKey.set(key, []);
+                        byKey.get(key).push(Number(r.predicted_max_level));
+                    });
+                    predSeries = sampled.map((row) => {
+                        const vals = byKey.get(keyOf(row.timestamp));
+                        return vals ? vals.reduce((a, c) => a + c, 0) / vals.length : null;
+                    });
+                    if (!predSeries.some((v) => v != null)) predSeries = null;
+                }
+            } catch (error) { /* overlay is optional */ }
+
             // Safely clear existing chart instance
             if (state.historyChart) { state.historyChart.destroy(); state.historyChart = null; }
             // Clear any orphan Chart.js instance on the canvas
@@ -645,7 +673,8 @@
                     labels: sampled.map((row) => new Date(row.timestamp).toLocaleString('en-TH', { timeZone: 'Asia/Bangkok', month: 'short', day: 'numeric', year: '2-digit', hour: '2-digit' })),
                     datasets: [
                         { label: 'Discharge (m³/s)', data: sampled.map((row) => row.discharge), borderColor: '#087da5', backgroundColor: 'rgba(8,125,165,.12)', yAxisID: 'flow', pointRadius: 0, tension: .25 },
-                        { label: 'Water level (m)', data: sampled.map((row) => row.water_level), borderColor: '#d99018', yAxisID: 'level', pointRadius: 0, tension: .25 }
+                        { label: 'Water level (m)', data: sampled.map((row) => row.water_level), borderColor: '#d99018', yAxisID: 'level', pointRadius: 0, tension: .25 },
+                        ...(predSeries ? [{ label: 'Model 24 h peak (as issued)', data: predSeries, borderColor: '#7c4fd0', borderDash: [6, 4], yAxisID: 'level', pointRadius: 0, tension: .25, spanGaps: true }] : [])
                     ]
                 },
                 options: {

src/web_api.py

@@ -60,7 +60,9 @@ HISTORY_TTL = 300  # 5 minutes
 FORECAST_CACHE: Dict[str, tuple] = {}
 FORECAST_CACHE_LOCK = Lock()
 FORECAST_COMPUTE_LOCK = asyncio.Lock()  # single-flight for expensive inference
-FORECAST_TTL = 900  # 15 minutes
+# The leader worker precomputes after every scrape cycle (hourly); the TTL just
+# needs to outlive one cycle so user requests never trigger inference themselves.
+FORECAST_TTL = int(os.getenv("FORECAST_TTL_SECONDS", "4500"))
 
 DB_STATS_CACHE: Dict[str, tuple] = {}
 DB_STATS_CACHE_LOCK = Lock()
@@ -159,6 +161,20 @@ async def lifespan(app: FastAPI):
     db_config = Config.get_database_config()
     app_state["scraper"] = EnhancedWaterMonitorScraper(db_config)
 
+    # Forecast history store (SQL only): records what the model predicted
+    try:
+        if db_config["type"] in ("sqlite", "postgresql", "mysql"):
+            from .forecast_history import ForecastHistoryStore
+
+            app_state["forecast_store"] = ForecastHistoryStore(
+                db_config["connection_string"], db_config["type"]
+            )
+        else:
+            app_state["forecast_store"] = None
+    except Exception as e:
+        app_state["forecast_store"] = None
+        logger.error(f"Forecast history store init failed: {e}")
+
     # Initialize HII/ThaiWater collector (rainfall + backup water level)
     try:
         from .hii_collector import create_collector_from_config
@@ -248,6 +264,32 @@ app.add_middleware(
 )
 
 
+async def _precompute_forecasts():
+    """Refresh the forecast cache and persist the issued forecasts (leader only)."""
+    try:
+        from .ml.predict import get_latest_forecasts
+    except ImportError:
+        return
+    try:
+        rows = await asyncio.to_thread(get_latest_forecasts)
+    except Exception as e:
+        logger.warning(f"Forecast precompute failed: {e}")
+        return
+    if not rows:
+        return
+    with FORECAST_CACHE_LOCK:
+        FORECAST_CACHE["all"] = (time.monotonic(), rows)
+    store = app_state.get("forecast_store")
+    if store:
+        try:
+            saved = await asyncio.to_thread(store.save_rows, rows)
+            logger.info(
+                f"Forecast precompute: cached {len(rows)} rows, persisted {saved}"
+            )
+        except Exception as e:
+            logger.warning(f"Forecast history save failed: {e}")
+
+
 async def background_scraping_task():
     """Background task for periodic data scraping"""
     while True:
@@ -309,6 +351,12 @@ async def background_scraping_task():
                         except Exception as e:
                             logger.error(f"HII collection failed: {e}")
 
+                    # Precompute forecasts on fresh data: primes the response
+                    # cache (user requests never pay for inference) and records
+                    # what the model predicted for later predicted-vs-actual
+                    # evaluation.
+                    await _precompute_forecasts()
+
                 app_state["is_scraping"] = False
 
             # Calculate next run time
@@ -366,6 +414,7 @@ def _send_umami_event(path: str, method: str, status: int, host: str, user_agent
 _CACHE_CONTROL_RULES = (
     ("/static/", "public, max-age=3600"),
     ("/measurements/latest", "public, max-age=30"),
+    ("/api/forecast/", "public, max-age=300"),
     ("/api/hii/", "public, max-age=60"),
     ("/measurements/history", "public, max-age=300"),
     ("/forecast", "public, max-age=120"),
@@ -1007,6 +1056,35 @@ async def _compute_forecasts(get_latest_forecasts):
     return data
 
 
+@app.get("/api/forecast/history/{station_code}")
+async def get_forecast_history(
+    station_code: str,
+    hours: int = Query(168, ge=1),
+    start: Optional[date] = Query(None, description="First day (overrides hours)"),
+    end: Optional[date] = Query(None, description="Last day, inclusive"),
+    horizon: Optional[int] = Query(None, ge=1, le=48),
+):
+    """Issued model forecasts for a station — what the model predicted, when.
+
+    Populated by the hourly precompute; enables predicted-vs-actual charts.
+    """
+    increment_counter("api_requests", labels={"endpoint": "forecast_history"})
+    store = app_state.get("forecast_store")
+    if not store:
+        return []
+    end_dt = (
+        datetime.combine(end, datetime.max.time()) if end else datetime.now()
+    )
+    start_dt = (
+        datetime.combine(start, datetime.min.time())
+        if start
+        else end_dt - timedelta(hours=hours)
+    )
+    return await asyncio.to_thread(
+        store.fetch, station_code, start_dt, end_dt, horizon
+    )
+
+
 @app.get("/measurements/latest", response_model=List[MeasurementResponse])
 async def get_latest_measurements(response: Response, limit: int = 100):
     """Get latest measurements from all stations"""

tests/test_dashboard.py

@@ -77,6 +77,8 @@ def test_dashboard_loads_station_history_chart():
     html = DASHBOARD_PATH.read_text(encoding="utf-8")
 
     assert "Station history" in html
+    assert "/api/forecast/history/" in html
+    assert "Model 24 h peak (as issued)" in html
     assert "PostgreSQL" not in html
     assert "/measurements/history/" in html
     assert "history-chart" in html

tests/test_forecast_history.py

@@ -0,0 +1,99 @@
+"""Tests for the issued-forecast archive (store + API endpoint)."""
+
+import asyncio
+import datetime
+
+import pytest
+
+from src.forecast_history import ForecastHistoryStore
+
+
+def _rows(as_of="2026-08-12T10:00:00"):
+    return [
+        {
+            "as_of": as_of,
+            "station_code": "P.1",
+            "horizon_hours": h,
+            "predicted_max_level": 2.8 + h / 100,
+            "p_warning": 0.02,
+            "p_danger": 0.001,
+            "current_level": 2.76,
+            "model_version": "hgb-v1+test",
+            "trained_at": "2026-08-12T10:52:00",
+            "source": "model",
+        }
+        for h in (6, 12, 24)
+    ]
+
+
+class TestForecastHistoryStore:
+    @pytest.fixture
+    def store(self, tmp_path):
+        store = ForecastHistoryStore(f"sqlite:///{tmp_path}/fh.db", "sqlite")
+        assert store.connect()
+        return store
+
+    def test_rejects_non_sql(self):
+        with pytest.raises(ValueError):
+            ForecastHistoryStore("http://x", "victoriametrics")
+
+    def test_roundtrip_and_upsert(self, store):
+        assert store.save_rows(_rows()) == 3
+        # Same as_of again -> upsert, still 3 rows
+        assert store.save_rows(_rows()) == 3
+        rows = store.fetch("P.1")
+        assert len(rows) == 3
+        assert [r["horizon_hours"] for r in rows] == [6, 12, 24]
+        assert rows[2]["predicted_max_level"] == pytest.approx(3.04)
+        assert rows[0]["model_version"] == "hgb-v1+test"
+
+    def test_fetch_filters(self, store):
+        store.save_rows(_rows("2026-08-12T10:00:00"))
+        store.save_rows(_rows("2026-08-12T11:00:00"))
+        only_24 = store.fetch("P.1", horizon_hours=24)
+        assert len(only_24) == 2
+        assert all(r["horizon_hours"] == 24 for r in only_24)
+        windowed = store.fetch(
+            "P.1",
+            start=datetime.datetime(2026, 8, 12, 10, 30),
+            end=datetime.datetime(2026, 8, 12, 12, 0),
+        )
+        assert len(windowed) == 3  # only the 11:00 issue
+        assert store.fetch("P.99") == []
+
+    def test_skips_malformed_rows(self, store):
+        rows = _rows() + [{"station_code": None, "as_of": None}]
+        assert store.save_rows(rows) == 3
+
+
+class TestForecastHistoryEndpoint:
+    def test_endpoint_returns_rows(self, tmp_path, monkeypatch):
+        from src import web_api
+
+        store = ForecastHistoryStore(f"sqlite:///{tmp_path}/api-fh.db", "sqlite")
+        assert store.connect()
+        now = datetime.datetime.now().replace(minute=0, second=0, microsecond=0)
+        store.save_rows(_rows(now.isoformat()))
+        monkeypatch.setitem(web_api.app_state, "forecast_store", store)
+
+        rows = asyncio.run(
+            web_api.get_forecast_history(
+                "P.1", hours=48, start=None, end=None, horizon=24
+            )
+        )
+        assert len(rows) == 1
+        assert rows[0]["horizon_hours"] == 24
+        assert rows[0]["station_code"] == "P.1"
+
+    def test_endpoint_without_store(self, monkeypatch):
+        from src import web_api
+
+        monkeypatch.setitem(web_api.app_state, "forecast_store", None)
+        assert (
+            asyncio.run(
+                web_api.get_forecast_history(
+                    "P.1", hours=168, start=None, end=None, horizon=None
+                )
+            )
+            == []
+        )