diff --git a/.gitignore b/.gitignore index 54632f6..a4e7db0 100644 --- a/.gitignore +++ b/.gitignore @@ -148,6 +148,9 @@ grafana_data/ models/*.joblib models/cache/ models/metrics.json +# scripts/retrain.sh working dirs (staging + one rollback generation) +models/.staging/ +models/.previous/ # Playwright MCP browser artifacts (screenshots/snapshots from agent sessions) .playwright-mcp/ diff --git a/README.md b/README.md index 7289d15..1803a07 100644 --- a/README.md +++ b/README.md @@ -288,9 +288,9 @@ before starting if the script reports it is missing. ```bash sudo useradd --system --no-create-home --shell /usr/sbin/nologin water-monitor -sudo cp scripts/water-monitor.service /etc/systemd/system/ -sudo systemctl enable water-monitor.service -sudo systemctl start water-monitor.service +uv sync --python 3.11 # creates .venv, the interpreter both units run +sudo cp scripts/water-monitor.service scripts/water-monitor-retrain.service scripts/water-monitor-retrain.timer /etc/systemd/system/ +sudo systemctl enable --now water-monitor.service water-monitor-retrain.timer ``` diff --git a/docs/FLOOD_FORECASTING.md b/docs/FLOOD_FORECASTING.md index 3ee2765..5622377 100644 --- a/docs/FLOOD_FORECASTING.md +++ b/docs/FLOOD_FORECASTING.md @@ -733,6 +733,32 @@ timestamp in every bundle. Both are echoed in every `/forecast` row, so you can tell from the API response alone which code produced a forecast and how old the model is. +**Scheduled retrain (since 2026-09-12).** `scripts/water-monitor-retrain.timer` +fires `water-monitor-retrain.service` on the 1st of every month at 03:30 server +time (`Persistent=true`, so a missed run catches up at boot). The unit runs +`scripts/retrain.sh` as the service user with `OMP_NUM_THREADS=4`, `Nice=15`: + +1. trains all stations into `models/.staging/` (the API keeps serving the old + bundles throughout); +2. refuses to promote unless `metrics.json` reports a `hgb-v3+` version and at + least 14 trained stations (exit 3, staging discarded, old models untouched); +3. renames the new bundles into `models/`, moving the previous generation to + `models/.previous/` for rollback. + +No API restart: `predict.py` reloads bundles by mtime on the next hourly +precompute. `systemctl list-timers water-monitor-retrain.timer` shows the next +run; `sudo systemctl start water-monitor-retrain.service` runs it now (after a +flood, say); `journalctl -u water-monitor-retrain` has the log. The installer +(`scripts/install.sh`) enables the timer. + +**Why the trainer refuses to run without rain (since 2026-09-12).** On +2026-09-01 the server retrain could not reach the Open-Meteo archive on a +checkout with no `models/cache/`, logged a warning, and quietly overwrote the +v3 bundles with gauge-only v2 ones — the 13-hour early warning on the 2024 flood +became an 18-hour late one and nothing on the dashboard said so. `train_all()` +now raises `RainUnavailableError` (CLI exit 2) in that situation. Gauge-only +bundles are still available, but only by asking for them: `--no-rain`. + ## 8. Operations runbook All commands assume the project virtualenv is active (`.venv` locally). @@ -786,10 +812,12 @@ that went quiet, or a bad backfill) rather than a modelling one. python -m pytest tests/test_flood_forecast.py -v ``` -Seven tests covering leakage, label alignment, the coverage gate, forward-fill and -staleness, a train/predict round trip, the heuristic fallback, and feature-name -stability. The whole suite runs in about 8 seconds, so there is no excuse for -skipping it before a deploy. +Tests cover leakage, label alignment, the coverage gate, forward-fill and +staleness, a train/predict round trip, the heuristic fallback, feature-name +stability, and the rain-downgrade guard (no rain series → `RainUnavailableError`, +nothing written; `--no-rain` → v2; rain present → v3 with the rain columns in +`feature_names`). The file runs in well under a minute, so there is no excuse +for skipping it before a deploy. **Understanding graceful degradation.** Three things can make a forecast row non-model-backed, and all of them are visible in the payload: diff --git a/scripts/install.sh b/scripts/install.sh index 9a7bd71..25878cc 100755 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -18,6 +18,7 @@ APP_DIR="${APP_DIR:-/opt/thailand-water-monitor}" SERVICE_USER="${SERVICE_USER:-water-monitor}" SERVICE_GROUP="${SERVICE_GROUP:-${SERVICE_USER}}" SERVICE_NAME="water-monitor.service" +RETRAIN_NAME="water-monitor-retrain" # Resolve the repo root (parent of this scripts/ directory). SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" @@ -72,11 +73,18 @@ if ! command -v uv >/dev/null 2>&1; then fi UV="$(command -v uv)" -log "Creating virtualenv at ${APP_DIR}/venv" +log "Syncing uv-managed virtualenv at ${APP_DIR}/.venv" cd "${APP_DIR}" -# Named 'venv' (not uv's default .venv) to match the systemd unit's ExecStart. -"${UV}" venv venv -"${UV}" pip install --python venv/bin/python -r requirements.txt +# ONE environment: uv sync owns .venv/ (from pyproject.toml + uv.lock, so the +# ML extras such as scikit-learn/joblib are present) and both systemd units +# run its interpreter directly. Never create a second env by another name -- +# a stale 'venv/' once coexisted here and broke manual retrains with +# ModuleNotFoundError while the service itself ran fine. +"${UV}" sync --python 3.11 --frozen +if [ -d "${APP_DIR}/venv" ]; then + warn "Removing stale ${APP_DIR}/venv (superseded by .venv)" + rm -rf "${APP_DIR}/venv" +fi # 4. Environment file ---------------------------------------------------------- if [ ! -f "${APP_DIR}/.env" ]; then @@ -100,11 +108,14 @@ if [ -f "${APP_DIR}/.env" ]; then chmod 0600 "${APP_DIR}/.env" fi -# 6. Install and enable the systemd unit -------------------------------------- -log "Installing systemd unit" +# 6. Install and enable the systemd units ------------------------------------- +log "Installing systemd units" install -m 0644 "${SCRIPT_DIR}/${SERVICE_NAME}" "/etc/systemd/system/${SERVICE_NAME}" +install -m 0644 "${SCRIPT_DIR}/${RETRAIN_NAME}.service" "/etc/systemd/system/${RETRAIN_NAME}.service" +install -m 0644 "${SCRIPT_DIR}/${RETRAIN_NAME}.timer" "/etc/systemd/system/${RETRAIN_NAME}.timer" systemctl daemon-reload systemctl enable "${SERVICE_NAME}" +systemctl enable --now "${RETRAIN_NAME}.timer" log "Done." echo @@ -112,3 +123,5 @@ echo "Next steps:" echo " sudo systemctl start ${SERVICE_NAME}" echo " systemctl status ${SERVICE_NAME}" echo " sudo journalctl -u ${SERVICE_NAME} -f" +echo " systemctl list-timers ${RETRAIN_NAME}.timer # monthly flood-model retrain" +echo " sudo systemctl start ${RETRAIN_NAME}.service # retrain now" diff --git a/scripts/retrain.sh b/scripts/retrain.sh new file mode 100644 index 0000000..072387e --- /dev/null +++ b/scripts/retrain.sh @@ -0,0 +1,90 @@ +#!/usr/bin/env bash +# +# Retrain the flood forecast models safely. Run by water-monitor-retrain.timer +# (monthly) or by hand: sudo systemctl start water-monitor-retrain.service +# +# Why a script rather than ExecStart=train_flood_model.py: +# * train.py writes each station's bundle straight into models/ over ~12 min, +# and the API's hourly precompute reloads bundles by mtime. Training into +# a staging dir and mv-ing (atomic on one filesystem) means the API never +# sees a half-written joblib file or a mixed old/new set. +# * A run that produced gauge-only (v2) bundles, or trained too few stations, +# must NOT replace the deployed models. train.py already aborts on a +# missing rain series; this script re-checks the written metrics anyway. +# * No API restart is needed: predict.py reloads changed bundles on the next +# precompute (every scrape cycle, hourly), so the new models are live +# within an hour. Restart manually if you want them live immediately. +# +# Exit codes: 0 ok, 2 training refused (see log), 3 verification failed. +set -euo pipefail + +APP_DIR="${APP_DIR:-/opt/thailand-water-monitor}" +PYTHON="${PYTHON:-${APP_DIR}/.venv/bin/python}" +MODELS_DIR="${APP_DIR}/models" +STAGE_DIR="${MODELS_DIR}/.staging" +# P.4A is NOT_TRAINABLE by design (17% fill); 15 of 16 is the normal outcome. +MIN_TRAINED="${MIN_TRAINED:-14}" +EXPECT_VERSION_PREFIX="${EXPECT_VERSION_PREFIX:-hgb-v3+}" + +log() { printf '%s retrain: %s\n' "$(date '+%Y-%m-%d %H:%M:%S')" "$*"; } + +cd "${APP_DIR}" +[ -x "${PYTHON}" ] || { log "no interpreter at ${PYTHON} (run uv sync)"; exit 3; } + +rm -rf "${STAGE_DIR}" +mkdir -p "${STAGE_DIR}" +log "training into ${STAGE_DIR} (python=${PYTHON}, OMP_NUM_THREADS=${OMP_NUM_THREADS:-unset})" + +# train_flood_model.py exits 2 on a missing rain series (RainUnavailableError) +# instead of silently writing v2 bundles -- propagate that unchanged. +set +e +"${PYTHON}" scripts/train_flood_model.py --stations all --models-dir "${STAGE_DIR}" "$@" +rc=$? +set -e +if [ "${rc}" -ne 0 ]; then + log "training failed (exit ${rc}); deployed models untouched" + rm -rf "${STAGE_DIR}" + exit "${rc}" +fi + +# Verify before promoting. Reads metrics.json from the stage dir. +VERSION="$("${PYTHON}" - "${STAGE_DIR}/metrics.json" <<'PY' +import json, sys +m = json.load(open(sys.argv[1])) +print(m["model_version"]) +PY +)" +TRAINED="$("${PYTHON}" - "${STAGE_DIR}/metrics.json" <<'PY' +import json, sys +m = json.load(open(sys.argv[1])) +print(sum(1 for s in m["stations"].values() if s.get("status") == "trained")) +PY +)" +log "staged model_version=${VERSION} trained_stations=${TRAINED}" + +case "${VERSION}" in + "${EXPECT_VERSION_PREFIX}"*) ;; + *) + log "REFUSING to deploy: version '${VERSION}' does not start with '${EXPECT_VERSION_PREFIX}'" + rm -rf "${STAGE_DIR}" + exit 3 + ;; +esac +if [ "${TRAINED}" -lt "${MIN_TRAINED}" ]; then + log "REFUSING to deploy: only ${TRAINED} stations trained (< ${MIN_TRAINED})" + rm -rf "${STAGE_DIR}" + exit 3 +fi + +# Promote: per-file rename is atomic; readers see either the old or the new +# bundle, never a partial one. Keep one previous generation for rollback. +mkdir -p "${MODELS_DIR}/.previous" +for f in "${STAGE_DIR}"/flood_*.joblib "${STAGE_DIR}/metrics.json"; do + name="$(basename "${f}")" + if [ -f "${MODELS_DIR}/${name}" ]; then + mv -f "${MODELS_DIR}/${name}" "${MODELS_DIR}/.previous/${name}" + fi + mv -f "${f}" "${MODELS_DIR}/${name}" +done +rm -rf "${STAGE_DIR}" +log "deployed ${VERSION} (${TRAINED} stations); previous generation in models/.previous. The API picks it up on its next hourly precompute." diff --git a/scripts/train_flood_model.py b/scripts/train_flood_model.py index 9e9c212..8c6ebee 100644 --- a/scripts/train_flood_model.py +++ b/scripts/train_flood_model.py @@ -11,7 +11,7 @@ import sys sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) -from src.ml.train import main +from src.ml.train import cli if __name__ == "__main__": - main() + raise SystemExit(cli()) diff --git a/scripts/water-monitor-retrain.service b/scripts/water-monitor-retrain.service new file mode 100644 index 0000000..bcb55fe --- /dev/null +++ b/scripts/water-monitor-retrain.service @@ -0,0 +1,39 @@ +[Unit] +Description=Retrain the Ping River flood forecast models +Documentation=https://git.b4l.co.th/B4L/Northern-Thailand-Ping-River-Monitor/-/blob/master/docs/FLOOD_FORECASTING.md +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +User=water-monitor +Group=water-monitor +WorkingDirectory=/opt/thailand-water-monitor +EnvironmentFile=/opt/thailand-water-monitor/.env +# Same interpreter as water-monitor.service -- the uv-managed .venv. +# scripts/retrain.sh trains into models/.staging, refuses to promote anything +# that is not a rain-enabled (hgb-v3) set covering the expected stations, then +# renames the bundles into place. The API reloads them on its next hourly +# precompute; no restart, so a failed run leaves the old models serving. +ExecStart=/bin/bash /opt/thailand-water-monitor/scripts/retrain.sh +# HistGradientBoosting is CPU-bound; cap threads so training cannot starve +# the API (docs/FLOOD_FORECASTING.md section 6 measured 4 as the sweet spot). +Environment=OMP_NUM_THREADS=4 +Environment=PYTHONPATH=/opt/thailand-water-monitor +Environment=PYTHONUNBUFFERED=1 +Nice=15 +IOSchedulingClass=idle +# 15 stations at ~50 s each plus data load: 12 min observed on 2026-09-12. +TimeoutStartSec=45min + +# Same sandbox as the API unit. +NoNewPrivileges=true +PrivateTmp=true +ProtectSystem=strict +ProtectHome=true +ReadWritePaths=/opt/thailand-water-monitor +CapabilityBoundingSet= + +StandardOutput=journal +StandardError=journal +SyslogIdentifier=water-monitor-retrain diff --git a/scripts/water-monitor-retrain.timer b/scripts/water-monitor-retrain.timer new file mode 100644 index 0000000..6606911 --- /dev/null +++ b/scripts/water-monitor-retrain.timer @@ -0,0 +1,17 @@ +[Unit] +Description=Monthly flood-model retrain (docs/FLOOD_FORECASTING.md section 7) + +[Timer] +# Policy: at minimum once pre-monsoon (May-June), monthly through the season +# (July-November), and after any major flood. A retrain costs ~12 min and RAM +# peaks ~300 MB, so running it every month all year is cheaper than remembering +# which months matter. 1st of the month, 03:30 server-local -- between the +# hourly scrapes and outside Thai daytime traffic. +OnCalendar=*-*-01 03:30:00 +# Catch up if the box was off at the scheduled time. +Persistent=true +RandomizedDelaySec=20min +Unit=water-monitor-retrain.service + +[Install] +WantedBy=timers.target diff --git a/scripts/water-monitor.service b/scripts/water-monitor.service index 28e3b2d..949707b 100644 --- a/scripts/water-monitor.service +++ b/scripts/water-monitor.service @@ -9,17 +9,19 @@ Type=simple User=water-monitor Group=water-monitor WorkingDirectory=/opt/thailand-water-monitor -ExecStart=/opt/thailand-water-monitor/venv/bin/python src/water_scraper_v3.py +# The uv-managed env (uv sync -> .venv). Same interpreter for water-monitor-retrain.service. +ExecStart=/opt/thailand-water-monitor/.venv/bin/python run.py --web-api ExecReload=/bin/kill -HUP $MAINPID Restart=always RestartSec=60 TimeoutStopSec=30 -# Environment variables -Environment=DB_TYPE=victoriametrics -Environment=VM_HOST=localhost -Environment=VM_PORT=8428 +# DB_TYPE / POSTGRES_CONNECTION_STRING / MATRIX_* come from the .env file. +EnvironmentFile=/opt/thailand-water-monitor/.env Environment=PYTHONPATH=/opt/thailand-water-monitor +# Serving path is latency-bound; single-threaded BLAS is 2.6x faster per call +# (docs/FLOOD_FORECASTING.md section 6). Training sets its own value. +Environment=OMP_NUM_THREADS=1 Environment=PYTHONUNBUFFERED=1 # Security settings diff --git a/src/ml/train.py b/src/ml/train.py index 682a97e..178405f 100644 --- a/src/ml/train.py +++ b/src/ml/train.py @@ -45,6 +45,14 @@ MIN_SIGMA = 0.15 MIN_ROWS_TO_TRAIN = 200 MIN_ROWS_FOR_HEAD = 50 + +class RainUnavailableError(RuntimeError): + """Raised when a rain-enabled training run cannot obtain the rain series. + + Training would otherwise fall through to gauge-only (v2) bundles and + overwrite the deployed v3 artifacts without anyone noticing. + """ + HGB_PARAMS = { "max_iter": 300, "learning_rate": 0.06, @@ -456,8 +464,13 @@ def train_all( models_dir = Path(models_dir) models_dir.mkdir(parents=True, exist_ok=True) - # Catchment rain (Open-Meteo archive, 2021+). Optional: without it the - # models train as v2 (no rain columns) and still serve correctly. + # Catchment rain (Open-Meteo archive, 2021+). A rain-less run produces v2 + # bundles that serve fine but have measurably less flood lead (the 2024 + # record flood: 13 h early with rain vs 18 h late without). The 2026-09-01 + # server retrain hit exactly that -- the archive fetch failed on a checkout + # with no models/cache/ and the run quietly wrote v2 over v3. So the + # downgrade is now an error unless the caller opts out with use_rain=False + # (the --no-rain flag), which is the only way to get v2 deliberately. rain_series = None if use_rain: try: @@ -465,7 +478,19 @@ def train_all( rain_series = rain_mod.catchment_mean(rain_mod.load_history()) except Exception as error: - logger.warning(f"rain history unavailable, training without it: {error}") + raise RainUnavailableError( + f"rain history unavailable ({error}); refusing to silently " + "downgrade to v2 bundles -- fix Open-Meteo access or restore " + "models/cache/rain_openmeteo.csv.gz, or pass --no-rain to " + "train gauge-only bundles on purpose" + ) from error + if rain_series is None: + raise RainUnavailableError( + "rain history unavailable (Open-Meteo archive unreachable and " + "no models/cache/rain_openmeteo.csv.gz); refusing to silently " + "downgrade to v2 bundles -- fix access, restore the cache file, " + "or pass --no-rain to train gauge-only bundles on purpose" + ) if rain_series is not None: logger.info( f"rain series: {rain_series.index.min()} .. {rain_series.index.max()}" @@ -584,7 +609,9 @@ def main(argv: Optional[List[str]] = None) -> None: parser.add_argument( "--no-rain", action="store_true", - help="train without the Open-Meteo rain features (v2-style bundles)", + help="DELIBERATELY train without the Open-Meteo rain features " + "(v2-style bundles). Without this flag a missing rain series aborts " + "the run instead of quietly downgrading the deployed model", ) parser.add_argument( "--dam", @@ -627,9 +654,21 @@ def main(argv: Optional[List[str]] = None) -> None: 1 for s in metrics_payload["stations"].values() if s["status"] == "trained" ) logger.info( - f"Done: {trained}/{len(stations)} stations trained. metrics.json written to {args.models_dir}" + f"Done: {trained}/{len(stations)} stations trained " + f"({metrics_payload['model_version']}). " + f"metrics.json written to {args.models_dir}" ) +def cli() -> int: + """Console entry: RainUnavailableError becomes a one-line error, exit 2.""" + try: + main() + except RainUnavailableError as error: + logger.error(str(error)) + return 2 + return 0 + + if __name__ == "__main__": - main() + raise SystemExit(cli()) diff --git a/tests/test_flood_forecast.py b/tests/test_flood_forecast.py index 4aa1aee..c4c3182 100644 --- a/tests/test_flood_forecast.py +++ b/tests/test_flood_forecast.py @@ -233,6 +233,75 @@ def test_heuristic_fallback(tmp_path): assert row["trained_at"] is None +def _p1_synth(n: int = 300, seed: int = 11) -> pd.DataFrame: + upstream = [code for code, _lead in features.UPSTREAM_LEADS["P.1"]] + return make_synth(n, ["P.1"] + upstream, seed=seed, pulses={"P.1": [(100, 20, 2.0)]}) + + +def test_train_refuses_silent_rain_downgrade(tmp_path, monkeypatch): + """use_rain=True with no rain series must abort, not write v2 bundles. + + Regression for the 2026-09-01 server retrain that overwrote v3 with v2 + because the Open-Meteo archive fetch failed on a cache-less checkout. + """ + from src.ml import rain as rain_mod + + df = _p1_synth() + overrides = {"max_iter": 10} + + # Case 1: the loader returns None (archive unreachable, no cache file) + monkeypatch.setattr(rain_mod, "load_history", lambda *a, **k: None) + with pytest.raises(train.RainUnavailableError, match="--no-rain"): + train.train_all(df, ["P.1"], models_dir=tmp_path, skip_eval=True, hgb_overrides=overrides, use_rain=True, use_dam=False) + assert not (tmp_path / "flood_P.1.joblib").exists() + assert not (tmp_path / "metrics.json").exists() + + # Case 2: the loader raises (network / parse error) + def boom(*a, **k): + raise ConnectionError("simulated Open-Meteo outage") + + monkeypatch.setattr(rain_mod, "load_history", boom) + with pytest.raises(train.RainUnavailableError, match="simulated Open-Meteo outage"): + train.train_all(df, ["P.1"], models_dir=tmp_path, skip_eval=True, hgb_overrides=overrides, use_rain=True, use_dam=False) + assert not (tmp_path / "flood_P.1.joblib").exists() + + # Explicit opt-out still produces v2 bundles as before + metrics = train.train_all(df, ["P.1"], models_dir=tmp_path, skip_eval=True, hgb_overrides=overrides, use_rain=False, use_dam=False) + assert metrics["model_version"].startswith("hgb-v2+") + assert (tmp_path / "flood_P.1.joblib").exists() + + +def test_train_with_rain_series_yields_v3(tmp_path, monkeypatch): + from src.ml import rain as rain_mod + + df = _p1_synth() + idx = pd.date_range(df["timestamp"].min(), df["timestamp"].max(), freq="h") + fake_rain = pd.DataFrame({"a": np.linspace(0, 1, len(idx)), "b": 0.5}, index=idx) + monkeypatch.setattr(rain_mod, "load_history", lambda *a, **k: fake_rain) + + metrics = train.train_all(df, ["P.1"], models_dir=tmp_path, skip_eval=True, hgb_overrides={"max_iter": 10}, use_rain=True, use_dam=False) + assert metrics["model_version"].startswith("hgb-v3+") + bundle = joblib.load(tmp_path / "flood_P.1.joblib") + assert set(features.RAIN_FEATURES) <= set(bundle["feature_names"]) + + +def test_cli_exit_code_on_rain_failure(tmp_path, monkeypatch, caplog): + """The console entry turns the guard into a one-line error and exit 2.""" + from src.ml import rain as rain_mod + + df = _p1_synth() + monkeypatch.setattr(rain_mod, "load_history", lambda *a, **k: None) + monkeypatch.setattr(train, "load_measurements", lambda *a, **k: df) + monkeypatch.setattr(train, "resolve_db_url", lambda *a, **k: None) + monkeypatch.setattr( + "sys.argv", + ["train", "--stations", "P.1", "--models-dir", str(tmp_path), "--skip-eval"], + ) + assert train.cli() == 2 + assert "refusing to silently downgrade" in caplog.text + assert not (tmp_path / "metrics.json").exists() + + def test_feature_name_stability(tmp_path): upstream = [code for code, _lead in features.UPSTREAM_LEADS["P.1"]] data_stations = ["P.1"] + upstream