feat: openmeteo_rain 2021+ backfill entry point

rain.backfill_db pushes the full cached Open-Meteo archive into the
openmeteo_rain table in 5k-row idempotent upsert chunks;
scripts/backfill_rain_db.py is the thin CLI (DB from Config/.env or
--db-url). Safe to re-run and safe alongside the hourly live writer.

scripts/backfill_rain_db.py

@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+"""Backfill the openmeteo_rain table with the full 2021+ catchment history.
+
+Usage:
+    uv run scripts/backfill_rain_db.py                 # DB from Config/.env
+    uv run scripts/backfill_rain_db.py --db-url postgresql://...
+"""
+
+import argparse
+import logging
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.config import Config
+from src.ml.rain import backfill_db
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--db-url", default=None)
+    args = parser.parse_args(argv)
+
+    logging.basicConfig(
+        level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
+    )
+    if args.db_url:
+        connection_string, db_type = args.db_url, args.db_url.split(":", 1)[0]
+    else:
+        cfg = Config.get_database_config()
+        if cfg["type"] not in ("sqlite", "postgresql", "mysql"):
+            print(f"requires a SQL database, got {cfg['type']}", file=sys.stderr)
+            return 1
+        connection_string, db_type = cfg["connection_string"], cfg["type"]
+
+    from sqlalchemy import create_engine
+
+    engine = create_engine(connection_string, pool_pre_ping=True)
+    saved = backfill_db(engine, db_type)
+    print(f"backfilled {saved} hourly rows into openmeteo_rain")
+    return 0 if saved else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/ml/rain.py

@@ -162,6 +162,24 @@ def serving_series() -> Optional[pd.Series]:
         return None
 
 
+def backfill_db(engine, db_type: str, chunk_rows: int = 5000) -> int:
+    """Push the full Open-Meteo history (2021+) into openmeteo_rain.
+
+    Loads (or fetches) the archive cache and upserts in chunks; idempotent,
+    safe to re-run, and safe alongside the hourly live writer.
+    """
+    history = load_history()
+    if history is None or history.empty:
+        logger.error("no rain history available to backfill")
+        return 0
+    total = 0
+    for start in range(0, len(history), chunk_rows):
+        part = history.iloc[start: start + chunk_rows]
+        total += save_to_db(part, engine, db_type)
+        logger.info(f"openmeteo_rain backfill: {total}/{len(history)} rows")
+    return total
+
+
 def save_to_db(df: pd.DataFrame, engine, db_type: str) -> int:
     """Upsert per-point + catchment-mean hourly rain into openmeteo_rain.