feat: backfill hii_waterlevel from the HII waterlevel_graph archive

scripts/backfill_hii_waterlevel.py walks the api-v3 waterlevel_graph
endpoint (hourly wl_msl + discharge, archive back to ~2019) in full-year
windows per station and upserts into hii_waterlevel. Defaults to the
RID-mirror and key stations; --stations/--all/--start/--end/--chunk-days
override. History upserts touch only wl_msl and discharge so colliding
live-snapshot rows keep storage_percent/situation_level. Idempotent and
safe to re-run.

docs/DATA_SOURCES.md

@@ -238,7 +238,7 @@ Oct 2024 flood). Mae Kuang Udom Thara is the second upstream reservoir.
 | ThaiWater `/v2/waterlevel` | 🟡 Display-only (`src/thaiwater.py`, needs `THAIWATER_API_KEY`, never persisted) | Optionally persist |
 | HII `rain_24h` | ✅ Ingested hourly via `src/hii_collector.py` → `hii_rain_stations` + `hii_rainfall` (Ping-filtered; ~300 stations) | — |
 | HII `waterlevel_load` | ✅ Ingested hourly via `src/hii_collector.py` → `hii_wl_stations` + `hii_waterlevel` (125 Ping stations, m MSL; `rid_code` column maps mirrors like `ridhydro_P.1` → `P.1`, `offset_msl` converts MSL → gauge datum) | — |
-| HII `waterlevel_graph` | ❌ Not ingested | Backfill script to cross-validate / gap-fill RID history (≥2019) |
+| HII `waterlevel_graph` | ✅ Backfill via `scripts/backfill_hii_waterlevel.py` → `hii_waterlevel` (hourly MSL + discharge, archive ≥2019; full-year windows per request; upserts never overwrite live-snapshot columns) | Run once on the box: `python scripts/backfill_hii_waterlevel.py` (defaults: 2019-01-01 → today, RID-mirror + key stations; `--stations P.1,P.67`, `--all` for every Ping station) |
 | Mae Ngat reservoir (lsim.rid.go.th) | ❌ Not ingested | Add daily storage/release scrape |
 | Satellite QPE (GSMaP/IMERG) | ❌ | Backfill training rainfall (GEE) |
 | Open-Meteo forecasts | ❌ | Add forecast features (live + 2021 archive for training) |

scripts/backfill_hii_waterlevel.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+"""CLI entry point for backfilling hii_waterlevel from the HII archive.
+
+Usage:
+    python scripts/backfill_hii_waterlevel.py                      # key stations, 2019..today
+    python scripts/backfill_hii_waterlevel.py --stations P.1,P.67
+    python scripts/backfill_hii_waterlevel.py --start 2024-09-01 --end 2024-11-01 --all
+"""
+
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.hii_backfill import main
+
+if __name__ == "__main__":
+    sys.exit(0 if main() else 1)

src/hii_backfill.py

@@ -0,0 +1,222 @@
+"""Backfill historical water levels from the HII waterlevel_graph endpoint.
+
+The api-v3 waterlevel_graph archive reaches back to ~2019 with hourly
+wl_msl + discharge. This module walks a date range in chunks per station and
+upserts into hii_waterlevel (idempotent; safe to re-run and to overlap with
+the live snapshot collector). Station metadata comes from a live
+waterlevel_load fetch, so hii_wl_stations is populated/refreshed as a side
+effect.
+
+Usage: python scripts/backfill_hii_waterlevel.py --start 2019-01-01
+"""
+
+import argparse
+import datetime
+import logging
+import time
+from typing import Dict, List, Optional
+
+from .hii_collector import (
+    HiiClient,
+    HiiStore,
+    PING_BASIN_CODE,
+    _parse_datetime,
+    _to_float,
+)
+
+logger = logging.getLogger(__name__)
+
+DEFAULT_START = datetime.date(2019, 1, 1)
+DEFAULT_CHUNK_DAYS = 365  # full-year windows verified working (8,760 rows, ~700KB)
+DEFAULT_SLEEP_SECONDS = 1.0
+
+
+def parse_graph_rows(payload: Dict) -> List[Dict]:
+    """Extract history rows from a waterlevel_graph payload (skips empty rows)."""
+    rows = (payload.get("data") or {}).get("graph_data") or []
+    records = []
+    for row in rows:
+        timestamp = _parse_datetime(row.get("datetime"))
+        wl_msl = _to_float(row.get("value"))
+        discharge = _to_float(row.get("discharge"))
+        if timestamp is None or (wl_msl is None and discharge is None):
+            continue
+        records.append(
+            {"timestamp": timestamp, "wl_msl": wl_msl, "discharge": discharge}
+        )
+    return records
+
+
+def fetch_waterlevel_history(
+    client: HiiClient,
+    station_id: int,
+    start_date: datetime.date,
+    end_date: datetime.date,
+) -> List[Dict]:
+    """Hourly wl_msl + discharge history (archive reaches back to ~2019)."""
+    payload = client.get(
+        "waterlevel_graph",
+        params={
+            "station_type": "tele_waterlevel",
+            "station_id": station_id,
+            "start_date": start_date.isoformat(),
+            "end_date": end_date.isoformat(),
+        },
+    )
+    return parse_graph_rows(payload)
+
+
+def chunk_date_range(
+    start: datetime.date, end: datetime.date, chunk_days: int
+) -> List[tuple]:
+    """Split [start, end] into inclusive (start, end) windows."""
+    chunks = []
+    cursor = start
+    while cursor <= end:
+        chunk_end = min(cursor + datetime.timedelta(days=chunk_days - 1), end)
+        chunks.append((cursor, chunk_end))
+        cursor = chunk_end + datetime.timedelta(days=1)
+    return chunks
+
+
+def select_stations(
+    station_records: List[Dict],
+    codes: Optional[List[str]] = None,
+    all_stations: bool = False,
+) -> List[Dict]:
+    """Pick stations to backfill from parsed waterlevel_load records.
+
+    Default: stations that mirror a RID gauge (rid_code) or are flagged
+    is_key_station — the ones relevant to the flood model. Explicit codes
+    match rid_code or oldcode; --all takes every station in the basin.
+    """
+    if all_stations:
+        return station_records
+    if codes:
+        wanted = {c.strip().upper() for c in codes if c.strip()}
+        return [
+            r
+            for r in station_records
+            if (r.get("rid_code") or "").upper() in wanted
+            or (r.get("oldcode") or "").upper() in wanted
+        ]
+    return [r for r in station_records if r.get("rid_code") or r.get("is_key_station")]
+
+
+def backfill(
+    store: HiiStore,
+    client: Optional[HiiClient] = None,
+    start: datetime.date = DEFAULT_START,
+    end: Optional[datetime.date] = None,
+    codes: Optional[List[str]] = None,
+    all_stations: bool = False,
+    chunk_days: int = DEFAULT_CHUNK_DAYS,
+    sleep_seconds: float = DEFAULT_SLEEP_SECONDS,
+    basin_code: int = PING_BASIN_CODE,
+) -> Dict[str, int]:
+    """Run the backfill; returns {'stations': n, 'rows': n, 'errors': n}."""
+    client = client or HiiClient()
+    end = end or datetime.date.today()
+
+    logger.info("Fetching station catalog from waterlevel_load...")
+    station_records = client.fetch_waterlevel(basin_code)
+    # Refresh station metadata (and today's snapshot) while we have it
+    store.save_waterlevel(station_records)
+
+    stations = select_stations(station_records, codes=codes, all_stations=all_stations)
+    if not stations:
+        logger.error("No stations matched the selection")
+        return {"stations": 0, "rows": 0, "errors": 0}
+
+    chunks = chunk_date_range(start, end, chunk_days)
+    logger.info(
+        f"Backfilling {len(stations)} stations x {len(chunks)} windows "
+        f"({start} .. {end}, {chunk_days}-day chunks)"
+    )
+
+    totals = {"stations": len(stations), "rows": 0, "errors": 0}
+    for station in stations:
+        sid = station["station_id"]
+        label = station.get("rid_code") or station.get("oldcode") or str(sid)
+        station_rows = 0
+        for chunk_start, chunk_end in chunks:
+            try:
+                rows = fetch_waterlevel_history(client, sid, chunk_start, chunk_end)
+                station_rows += store.save_waterlevel_history(sid, rows)
+            except Exception as e:
+                totals["errors"] += 1
+                logger.warning(
+                    f"{label}: {chunk_start}..{chunk_end} failed: {e}"
+                )
+            time.sleep(sleep_seconds)
+        totals["rows"] += station_rows
+        logger.info(f"{label} (id {sid}): {station_rows} rows saved")
+
+    logger.info(
+        f"Backfill complete: {totals['rows']} rows across "
+        f"{totals['stations']} stations, {totals['errors']} failed windows"
+    )
+    return totals
+
+
+def main(argv: Optional[List[str]] = None) -> bool:
+    parser = argparse.ArgumentParser(
+        description="Backfill hii_waterlevel from the HII waterlevel_graph archive"
+    )
+    parser.add_argument(
+        "--start",
+        type=datetime.date.fromisoformat,
+        default=DEFAULT_START,
+        help=f"First date to fetch (default {DEFAULT_START})",
+    )
+    parser.add_argument(
+        "--end",
+        type=datetime.date.fromisoformat,
+        default=None,
+        help="Last date to fetch (default today)",
+    )
+    parser.add_argument(
+        "--stations",
+        help="Comma-separated codes (rid_code or oldcode, e.g. P.1,P.67,CHM004). "
+        "Default: all RID-mirror and key stations",
+    )
+    parser.add_argument(
+        "--all",
+        action="store_true",
+        help="Backfill every Ping-basin station (125+; slow)",
+    )
+    parser.add_argument(
+        "--chunk-days", type=int, default=DEFAULT_CHUNK_DAYS, help="Window size"
+    )
+    parser.add_argument(
+        "--sleep",
+        type=float,
+        default=DEFAULT_SLEEP_SECONDS,
+        help="Pause between requests in seconds",
+    )
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(
+        level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+    )
+
+    from .config import Config
+
+    db_config = Config.get_database_config()
+    if db_config["type"] not in ("sqlite", "postgresql", "mysql"):
+        logger.error(f"Backfill requires a SQL DB_TYPE, got '{db_config['type']}'")
+        return False
+    store = HiiStore(db_config["connection_string"], db_config["type"])
+    if not store.connect():
+        return False
+
+    totals = backfill(
+        store,
+        start=args.start,
+        end=args.end,
+        codes=args.stations.split(",") if args.stations else None,
+        all_stations=args.all,
+        chunk_days=args.chunk_days,
+        sleep_seconds=args.sleep,
+    )
+    return totals["rows"] > 0 and totals["errors"] == 0

src/hii_collector.py

@@ -162,18 +162,18 @@ class HiiClient:
         self.session = session or requests.Session()
         self.timeout = timeout
 
-    def _get(self, endpoint: str) -> Dict:
+    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
         response = self.session.get(
-            f"{self.base_url}/{endpoint}", timeout=self.timeout
+            f"{self.base_url}/{endpoint}", params=params, timeout=self.timeout
         )
         response.raise_for_status()
         return response.json()
 
     def fetch_rain(self, basin_code: int = PING_BASIN_CODE) -> List[Dict]:
-        return parse_rain_records(self._get("rain_24h"), basin_code)
+        return parse_rain_records(self.get("rain_24h"), basin_code)
 
     def fetch_waterlevel(self, basin_code: int = PING_BASIN_CODE) -> List[Dict]:
-        return parse_waterlevel_records(self._get("waterlevel_load"), basin_code)
+        return parse_waterlevel_records(self.get("waterlevel_load"), basin_code)
 
 
 class HiiStore:
@@ -350,6 +350,41 @@ class HiiStore:
             ],
         )
 
+    def save_waterlevel_history(self, station_id: int, rows: List[Dict]) -> int:
+        """Upsert backfilled history rows, touching only wl_msl and discharge.
+
+        Live-snapshot rows for the same (station, hour) keep their extra
+        columns (storage_percent, situation_level, ...) untouched.
+        """
+        if not rows:
+            return 0
+        if not self.engine and not self.connect():
+            return 0
+        from sqlalchemy import text
+
+        cols = "(station_id, timestamp, wl_msl, discharge)"
+        values = "(:station_id, :timestamp, :wl_msl, :discharge)"
+        if self.db_type == "mysql":
+            sql = (
+                f"INSERT INTO hii_waterlevel {cols} VALUES {values} "
+                "ON DUPLICATE KEY UPDATE wl_msl = VALUES(wl_msl), "
+                "discharge = VALUES(discharge)"
+            )
+        else:  # sqlite (>=3.24) and postgresql share upsert syntax
+            sql = (
+                f"INSERT INTO hii_waterlevel {cols} VALUES {values} "
+                "ON CONFLICT (station_id, timestamp) DO UPDATE SET "
+                "wl_msl = EXCLUDED.wl_msl, discharge = EXCLUDED.discharge"
+            )
+        params = [{**row, "station_id": station_id} for row in rows]
+        try:
+            with self.engine.begin() as conn:
+                conn.execute(text(sql), params)
+            return len(params)
+        except Exception as e:
+            logger.error(f"HiiStore history save failed: {e}")
+            return 0
+
     def _save(
         self,
         records: List[Dict],

tests/test_hii_collector.py

@@ -4,6 +4,7 @@ import datetime
 
 import pytest
 
+from src.hii_backfill import chunk_date_range, parse_graph_rows, select_stations
 from src.hii_collector import (
     HiiStore,
     parse_rain_records,
@@ -207,3 +208,105 @@ class TestHiiStore:
 
     def test_save_empty(self, store):
         assert store.save_rain([]) == 0
+
+    def test_history_upsert_preserves_snapshot_columns(self, store):
+        # A live snapshot row exists with extra columns populated
+        store.save_waterlevel(parse_waterlevel_records(_waterlevel_payload()))
+        # Backfill collides on the same (station, timestamp) with new values
+        rows = [
+            {
+                "timestamp": datetime.datetime(2026, 8, 11, 13, 0),
+                "wl_msl": 303.30,
+                "discharge": 340.0,
+            },
+            {
+                "timestamp": datetime.datetime(2019, 8, 1, 1, 0),
+                "wl_msl": 301.71,
+                "discharge": 11.7,
+            },
+        ]
+        assert store.save_waterlevel_history(3226, rows) == 2
+
+        from sqlalchemy import text
+
+        with store.engine.connect() as conn:
+            collided = conn.execute(
+                text(
+                    "SELECT wl_msl, discharge, storage_percent, situation_level "
+                    "FROM hii_waterlevel WHERE station_id = 3226 "
+                    "AND timestamp = '2026-08-11 13:00:00'"
+                )
+            ).fetchone()
+            historical = conn.execute(
+                text(
+                    "SELECT wl_msl FROM hii_waterlevel WHERE station_id = 3226 "
+                    "AND timestamp = '2019-08-01 01:00:00'"
+                )
+            ).fetchone()
+        # wl_msl/discharge updated, snapshot-only columns untouched
+        assert float(collided[0]) == 303.30
+        assert float(collided[1]) == 340.0
+        assert float(collided[2]) == 81.21
+        assert collided[3] == 4
+        assert float(historical[0]) == 301.71
+
+
+class TestParseGraphRows:
+    def test_parses_and_skips_empty(self):
+        payload = {
+            "data": {
+                "graph_data": [
+                    {"datetime": "2024-10-05 12:00", "value": 305.8, "discharge": 656},
+                    {"datetime": "2024-10-05 13:00", "value": None, "discharge": None},
+                    {"datetime": None, "value": 300.0, "discharge": 1},
+                ]
+            }
+        }
+        rows = parse_graph_rows(payload)
+        assert rows == [
+            {
+                "timestamp": datetime.datetime(2024, 10, 5, 12, 0),
+                "wl_msl": 305.8,
+                "discharge": 656.0,
+            }
+        ]
+
+    def test_empty_payload(self):
+        assert parse_graph_rows({}) == []
+
+
+class TestBackfillHelpers:
+    def test_chunk_date_range(self):
+        chunks = chunk_date_range(
+            datetime.date(2024, 1, 1), datetime.date(2024, 3, 1), 31
+        )
+        assert chunks[0] == (datetime.date(2024, 1, 1), datetime.date(2024, 1, 31))
+        assert chunks[-1][1] == datetime.date(2024, 3, 1)
+        # Contiguous, no overlap
+        for (_, prev_end), (next_start, _) in zip(chunks, chunks[1:]):
+            assert next_start == prev_end + datetime.timedelta(days=1)
+
+    def test_chunk_single_day(self):
+        d = datetime.date(2024, 1, 1)
+        assert chunk_date_range(d, d, 365) == [(d, d)]
+
+    def test_select_default_keeps_rid_and_key_stations(self):
+        records = [
+            {"station_id": 1, "rid_code": "P.1", "is_key_station": True},
+            {"station_id": 2, "rid_code": None, "is_key_station": False},
+            {"station_id": 3, "rid_code": None, "is_key_station": True},
+        ]
+        assert [r["station_id"] for r in select_stations(records)] == [1, 3]
+
+    def test_select_by_code_matches_rid_code_and_oldcode(self):
+        records = [
+            {"station_id": 1, "rid_code": "P.1", "oldcode": "ridhydro_P.1"},
+            {"station_id": 2, "rid_code": None, "oldcode": "CHM004"},
+            {"station_id": 3, "rid_code": "P.67", "oldcode": "P.67"},
+        ]
+        selected = select_stations(records, codes=["p.1", "chm004"])
+        assert [r["station_id"] for r in selected] == [1, 2]
+
+    def test_select_all(self):
+        records = [{"station_id": 1}, {"station_id": 2}]
+        assert select_stations(records, all_stations=True) == records