From ba2db42eaa698e6ce2ac7b43568b546011cbdf43 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 23 Jun 2026 18:16:46 +0800 Subject: [PATCH] Restore DEB training settlement worker --- deploy.sh | 3 + docker-compose.yml | 43 ++++++ ...26-06-23-deb-training-settlement-worker.md | 138 +++++++++++++++++ tests/test_deployment_runtime_config.py | 21 +++ tests/test_training_settlement_service.py | 77 ++++++++++ tests/test_web_observability.py | 87 ++++++++++- web/diagnostics/health.py | 38 ++++- web/services/system_api.py | 50 ++++++ web/training_settlement_service.py | 142 ++++++++++++++++++ web/training_settlement_worker.py | 103 +++++++++++++ 10 files changed, 700 insertions(+), 2 deletions(-) create mode 100644 docs/superpowers/plans/2026-06-23-deb-training-settlement-worker.md create mode 100644 tests/test_training_settlement_service.py create mode 100644 web/training_settlement_service.py create mode 100644 web/training_settlement_worker.py diff --git a/deploy.sh b/deploy.sh index 7bb25498..823a624a 100644 --- a/deploy.sh +++ b/deploy.sh @@ -350,6 +350,9 @@ compose_up_retry "observation collector" -d --no-deps polyweather_collector echo "Updating cache warmer..." compose_up_retry "cache warmer" -d --no-deps polyweather_warmer +echo "Updating training settlement worker..." +compose_up_retry "training settlement" -d --no-deps polyweather_training_settlement + echo "Updating frontend..." compose_up_retry "frontend" -d --no-deps polyweather_frontend diff --git a/docker-compose.yml b/docker-compose.yml index 05c740c6..5fef525a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -253,6 +253,49 @@ services: volumes: - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data + polyweather_training_settlement: + command: python -m web.training_settlement_worker + container_name: polyweather_training_settlement + depends_on: + polyweather_redis: + condition: service_healthy + polyweather_web: + condition: service_started + logging: + driver: "json-file" + options: + max-size: "50m" + max-file: "3" + cpus: ${POLYWEATHER_TRAINING_SETTLEMENT_CPUS:-0.50} + env_file: *id001 + environment: + POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis} + POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false' + POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} + POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' + POLYWEATHER_SERVICE_ROLE: training_settlement + POLYWEATHER_TRAINING_SETTLEMENT_INITIAL_DELAY_SEC: ${POLYWEATHER_TRAINING_SETTLEMENT_INITIAL_DELAY_SEC:-60} + POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC: ${POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC:-21600} + POLYWEATHER_TRAINING_SETTLEMENT_LOOKBACK_DAYS: ${POLYWEATHER_TRAINING_SETTLEMENT_LOOKBACK_DAYS:-10} + healthcheck: + interval: 60s + retries: 3 + test: + - CMD + - python + - -c + - import sqlite3; c=sqlite3.connect('/var/lib/polyweather/polyweather.db'); + c.execute('SELECT 1'); c.close() + timeout: 10s + image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} + mem_limit: ${POLYWEATHER_TRAINING_SETTLEMENT_MEM_LIMIT:-768m} + memswap_limit: ${POLYWEATHER_TRAINING_SETTLEMENT_MEMSWAP_LIMIT:-1g} + pids_limit: 256 + restart: unless-stopped + user: ${UID:-1000}:${GID:-1000} + volumes: + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data x-polyweather-base: env_file: *id001 image: ghcr.io/yangyuan-zhen/polyweather-backend:${IMAGE_TAG:-latest} diff --git a/docs/superpowers/plans/2026-06-23-deb-training-settlement-worker.md b/docs/superpowers/plans/2026-06-23-deb-training-settlement-worker.md new file mode 100644 index 00000000..4ea70d97 --- /dev/null +++ b/docs/superpowers/plans/2026-06-23-deb-training-settlement-worker.md @@ -0,0 +1,138 @@ +# DEB Training Settlement Worker Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Restore automatic DEB training data freshness after the realtime collector/canonical-cache split. + +**Architecture:** Add a dedicated low-frequency training settlement service that runs city analysis for forecast/DEB snapshots and reconciles recent settled actual highs. Keep it separate from the high-frequency observation collector so user-facing realtime updates stay light. + +**Tech Stack:** Python, FastAPI service modules, Docker Compose, pytest, existing `update_daily_record()` and `reconcile_recent_actual_highs()` paths. + +--- + +### Task 1: Training Settlement Service + +**Files:** +- Create: `web/training_settlement_service.py` +- Create: `tests/test_training_settlement_service.py` + +- [ ] **Step 1: Write the failing service test** + +```python +def test_training_settlement_cycle_runs_analysis_and_reconciles_supported_cities(): + calls = {"analysis": [], "reconcile": []} + + def analysis_runner(city): + calls["analysis"].append(city) + return {"city": city, "deb": {"prediction": 31.2}} + + def reconciler(city, *, lookback_days): + calls["reconcile"].append((city, lookback_days)) + return {"ok": True, "updated": 1} + + result = run_training_settlement_cycle( + city_registry={ + "shanghai": {"icao": "ZSSS", "settlement_source": "metar"}, + "legacy": {"settlement_source": "wunderground"}, + }, + analysis_runner=analysis_runner, + actual_reconciler=reconciler, + lookback_days=9, + ) + + assert result["ok"] is True + assert result["processed"] == 1 + assert calls["analysis"] == ["shanghai"] + assert calls["reconcile"] == [("shanghai", 9)] +``` + +- [ ] **Step 2: Run the test to verify it fails** + +Run: `python -m pytest tests/test_training_settlement_service.py::test_training_settlement_cycle_runs_analysis_and_reconciles_supported_cities -q` + +Expected: FAIL because `web.training_settlement_service` does not exist. + +- [ ] **Step 3: Implement the service** + +Create `run_training_settlement_cycle()` with injectable `city_registry`, `analysis_runner`, and `actual_reconciler`. Default analysis runner calls `web.analysis_service._analyze(city, force_refresh=False, detail_mode="panel")`; default reconciler calls `src.analysis.deb_algorithm.reconcile_recent_actual_highs()`. + +- [ ] **Step 4: Run the test to verify it passes** + +Run: `python -m pytest tests/test_training_settlement_service.py -q` + +Expected: PASS. + +### Task 2: Worker Entrypoint And Compose + +**Files:** +- Create: `web/training_settlement_worker.py` +- Modify: `docker-compose.yml` +- Modify: `tests/test_deployment_runtime_config.py` + +- [ ] **Step 1: Write failing deployment tests** + +Assert the compose file contains `polyweather_training_settlement`, command `python -m web.training_settlement_worker`, role `training_settlement`, and conservative interval/lookback environment variables. + +- [ ] **Step 2: Run the deployment test to verify it fails** + +Run: `python -m pytest tests/test_deployment_runtime_config.py::test_runtime_compose_splits_realtime_workers -q` + +Expected: FAIL because the worker service is absent. + +- [ ] **Step 3: Implement worker and compose service** + +The worker loops `run_training_settlement_cycle()` every `POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC` seconds, with initial delay and lookback from environment. + +- [ ] **Step 4: Run focused tests** + +Run: `python -m pytest tests/test_training_settlement_service.py tests/test_deployment_runtime_config.py -q` + +Expected: PASS. + +### Task 3: Stale Monitoring + +**Files:** +- Modify: `web/diagnostics/health.py` +- Modify: `web/services/system_api.py` +- Modify: `tests/test_web_observability.py` + +- [ ] **Step 1: Write failing observability tests** + +Assert system status training summaries include `stale_days` for daily/truth/features, and Prometheus exports stale gauges for training data. + +- [ ] **Step 2: Run focused observability tests to verify failure** + +Run: `python -m pytest tests/test_web_observability.py::test_system_status_includes_training_data tests/test_web_observability.py::test_metrics_endpoint_returns_prometheus_payload_for_ops_admin -q` + +Expected: FAIL because stale fields/gauges are absent. + +- [ ] **Step 3: Implement stale summary and gauges** + +Add date-diff calculation against UTC today. Export `polyweather_daily_records_stale_days`, `polyweather_truth_records_stale_days`, `polyweather_training_features_stale_days`, and `polyweather_training_data_stale`. + +- [ ] **Step 4: Run focused observability tests** + +Run: `python -m pytest tests/test_web_observability.py::test_system_status_includes_training_data tests/test_web_observability.py::test_metrics_endpoint_returns_prometheus_payload_for_ops_admin -q` + +Expected: PASS. + +### Task 4: Verification And Backfill + +**Files:** +- Optional create: `scripts/backfill_training_settlement.py` + +- [ ] **Step 1: Run backend checks** + +Run: `python -m ruff check .` + +Run: `python -m pytest tests/test_training_settlement_service.py tests/test_deployment_runtime_config.py tests/test_web_observability.py -q` + +- [ ] **Step 2: Run local one-shot cycle** + +Run: `python -m web.training_settlement_worker --once --lookback-days 10 --cities shanghai` + +Expected: JSON-like log output with `ok=True`; local DB should get a fresh row for the current local target date if analysis succeeds. + +- [ ] **Step 3: Production note** + +Missed forecast snapshots from 2026-06-15 to 2026-06-22 cannot be reconstructed honestly unless archived city analysis payloads exist. The worker restores forward automatic samples; actual-high truth can still be reconciled for supported settlement sources. diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index cb3cd3cc..45060ead 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -94,6 +94,13 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): 1, )[0] warmer_block = compose.split(" polyweather_warmer:", 1)[1].split( + "\n polyweather_training_settlement:", + 1, + )[0] + training_settlement_block = compose.split( + " polyweather_training_settlement:", + 1, + )[1].split( "\nx-polyweather-base:", 1, )[0] @@ -102,6 +109,7 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): assert "POLYWEATHER_SERVICE_ROLE: bot" in compose assert "POLYWEATHER_SERVICE_ROLE: collector" in collector_block assert "POLYWEATHER_SERVICE_ROLE: warmer" in warmer_block + assert "POLYWEATHER_SERVICE_ROLE: training_settlement" in training_settlement_block assert "redis-server --appendonly yes --maxmemory ${POLYWEATHER_REDIS_MAXMEMORY:-512mb} --maxmemory-policy noeviction" in compose assert "POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false'" in bot_block assert "POLYWEATHER_EVENT_STORE: ${POLYWEATHER_EVENT_STORE:-redis}" in web_block @@ -117,6 +125,18 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in web_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'true'" in collector_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in warmer_block + assert "POLYWEATHER_OBSERVATION_COLLECTOR_ENABLED: 'false'" in training_settlement_block + assert "command: python -m web.training_settlement_worker" in training_settlement_block + assert ( + "POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC: " + "${POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC:-21600}" + in training_settlement_block + ) + assert ( + "POLYWEATHER_TRAINING_SETTLEMENT_LOOKBACK_DAYS: " + "${POLYWEATHER_TRAINING_SETTLEMENT_LOOKBACK_DAYS:-10}" + in training_settlement_block + ) assert "POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY:-3}" in web_block assert "POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY: ${POLYWEATHER_CITY_DETAIL_BATCH_GLOBAL_CONCURRENCY:-3}" in web_block assert "POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS: ${POLYWEATHER_CITY_DETAIL_BATCH_QUEUE_WAIT_MS:-3000}" in web_block @@ -287,6 +307,7 @@ def test_deploy_script_retries_compose_recreate_races(): assert 'compose_up_retry "backend services" -d --no-deps polyweather_web polyweather' in script assert 'compose_up_retry "observation collector" -d --no-deps polyweather_collector' in script assert 'compose_up_retry "cache warmer" -d --no-deps polyweather_warmer' in script + assert 'compose_up_retry "training settlement" -d --no-deps polyweather_training_settlement' in script assert 'compose_up_retry "frontend" -d --no-deps polyweather_frontend' in script diff --git a/tests/test_training_settlement_service.py b/tests/test_training_settlement_service.py new file mode 100644 index 00000000..075ecf16 --- /dev/null +++ b/tests/test_training_settlement_service.py @@ -0,0 +1,77 @@ +from web.training_settlement_service import run_training_settlement_cycle + + +def test_training_settlement_cycle_runs_analysis_and_reconciles_supported_cities(): + calls = {"analysis": [], "reconcile": []} + + def analysis_runner(city): + calls["analysis"].append(city) + return {"city": city, "deb": {"prediction": 31.2}} + + def actual_reconciler(city, *, lookback_days): + calls["reconcile"].append((city, lookback_days)) + return {"ok": True, "updated": 1} + + result = run_training_settlement_cycle( + city_registry={ + "shanghai": {"icao": "ZSSS", "settlement_source": "metar"}, + "legacy": {"settlement_source": "wunderground"}, + }, + analysis_runner=analysis_runner, + actual_reconciler=actual_reconciler, + lookback_days=9, + ) + + assert result["ok"] is True + assert result["processed"] == 1 + assert result["failed"] == 0 + assert result["unsupported"] == 1 + assert calls["analysis"] == ["shanghai"] + assert calls["reconcile"] == [("shanghai", 9)] + + +def test_training_settlement_cycle_continues_after_city_failure(): + calls = [] + + def analysis_runner(city): + calls.append(city) + if city == "shanghai": + raise RuntimeError("analysis unavailable") + return {"city": city} + + result = run_training_settlement_cycle( + city_registry={ + "shanghai": {"icao": "ZSSS", "settlement_source": "metar"}, + "tokyo": {"icao": "RJTT", "settlement_source": "metar"}, + }, + analysis_runner=analysis_runner, + actual_reconciler=lambda city, *, lookback_days: {"ok": True, "updated": 0}, + lookback_days=3, + ) + + assert result["ok"] is False + assert result["processed"] == 1 + assert result["failed"] == 1 + assert calls == ["shanghai", "tokyo"] + assert result["items"][0]["ok"] is False + assert result["items"][1]["ok"] is True + + +def test_training_settlement_cycle_skips_reconcile_for_non_reconcile_sources(): + calls = {"analysis": [], "reconcile": []} + + result = run_training_settlement_cycle( + city_registry={ + "taipei": {"icao": "RCSS", "settlement_source": "cwa"}, + }, + analysis_runner=lambda city: calls["analysis"].append(city) or {"city": city}, + actual_reconciler=lambda city, *, lookback_days: calls["reconcile"].append(city) + or {"ok": True}, + lookback_days=3, + ) + + assert result["ok"] is True + assert result["processed"] == 1 + assert calls["analysis"] == ["taipei"] + assert calls["reconcile"] == [] + assert result["items"][0]["reconcile"]["reason"] == "unsupported_reconcile_source" diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index fb1b5470..869bd224 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1,5 +1,6 @@ from datetime import datetime, timedelta, timezone +from types import SimpleNamespace from fastapi.testclient import TestClient from starlette.requests import Request @@ -9,13 +10,16 @@ import web.services.auth_api as auth_api from web.app import app import web.routes as routes import web.services.ops_api as ops_api +import web.diagnostics.health as diagnostics_health import web.scan_terminal_cache as scan_terminal_cache import web.scan_terminal_service as scan_terminal_service +import web.services.system_api as system_api import web.services.city_api as city_api import web.services.city_runtime as city_runtime from web.services.observation_freshness import build_observation_freshness from web.scan_terminal_cache import scan_terminal_cache_key -from src.database.runtime_state import TruthRecordRepository +from src.database.runtime_state import RuntimeStateDB, TruthRecordRepository +from src.utils.metrics import export_prometheus_metrics client = TestClient(app) @@ -83,6 +87,8 @@ def test_system_status_returns_summary_shape_for_ops_admin(monkeypatch): assert 'sse_connections' in payload['realtime'] assert 'truth_records' in payload['training_data'] assert 'training_features' in payload['training_data'] + assert 'stale_days' in payload['training_data']['truth_records'] + assert 'stale_days' in payload['training_data']['training_features'] assert 'city_coverage' in payload['training_data'] assert 'model_city_coverage' in payload['training_data'] assert 'metar_entries' in payload['cache'] @@ -121,6 +127,85 @@ def test_metrics_endpoint_returns_prometheus_payload_for_ops_admin(monkeypatch): assert 'polyweather_http_requests_total' in response.text +def test_training_data_summary_reports_stale_days(tmp_path): + db = RuntimeStateDB(str(tmp_path / "training.db")) + yesterday = (datetime.now(timezone.utc).date() - timedelta(days=1)).strftime("%Y-%m-%d") + stale_day = (datetime.now(timezone.utc).date() - timedelta(days=5)).strftime("%Y-%m-%d") + with db.connect() as conn: + conn.execute( + """ + INSERT INTO truth_records_store ( + city, target_date, actual_high, settlement_source, updated_at, is_final + ) VALUES ('shanghai', ?, 31.2, 'metar', 1, 1) + """, + (yesterday,), + ) + conn.execute( + """ + INSERT INTO training_feature_records_store ( + city, target_date, updated_at, payload_json + ) VALUES ('shanghai', ?, 1, '{}') + """, + (stale_day,), + ) + conn.commit() + + payload = diagnostics_health._training_data_summary( + SimpleNamespace(db_path=db.db_path), + {"shanghai": {"name": "Shanghai", "settlement_source": "metar", "icao": "ZSSS"}}, + ) + + assert payload["truth_records"]["stale_days"] == 1 + assert payload["training_features"]["stale_days"] == 5 + assert payload["stale"] is True + + +def test_prometheus_exports_training_data_stale_metrics(monkeypatch, tmp_path): + db = RuntimeStateDB(str(tmp_path / "training-metrics.db")) + stale_day = (datetime.now(timezone.utc).date() - timedelta(days=4)).strftime("%Y-%m-%d") + with db.connect() as conn: + conn.execute( + """ + INSERT INTO daily_records_store ( + city, target_date, actual_high, deb_prediction, mu, updated_at, payload_json + ) VALUES ('shanghai', ?, 31.2, 31.0, 31.1, 1, '{}') + """, + (stale_day,), + ) + conn.execute( + """ + INSERT INTO truth_records_store ( + city, target_date, actual_high, settlement_source, updated_at, is_final + ) VALUES ('shanghai', ?, 31.2, 'metar', 1, 1) + """, + (stale_day,), + ) + conn.execute( + """ + INSERT INTO training_feature_records_store ( + city, target_date, updated_at, payload_json + ) VALUES ('shanghai', ?, 1, '{}') + """, + (stale_day,), + ) + conn.commit() + + fake_db = SimpleNamespace( + db_path=db.db_path, + list_payment_audit_events=lambda **_kwargs: [], + list_refund_cases=lambda **_kwargs: [], + ) + monkeypatch.setattr(system_api, "DBManager", lambda: fake_db) + + system_api._refresh_operational_metrics() + metrics = export_prometheus_metrics() + + assert "polyweather_daily_records_stale_days 4" in metrics + assert "polyweather_truth_records_stale_days 4" in metrics + assert "polyweather_training_features_stale_days 4" in metrics + assert "polyweather_training_data_stale 1" in metrics + + def test_system_cache_status_requires_ops_admin(monkeypatch): monkeypatch.setattr(routes, "_assert_entitlement", lambda request: None) diff --git a/web/diagnostics/health.py b/web/diagnostics/health.py index 0822914d..a96cb778 100644 --- a/web/diagnostics/health.py +++ b/web/diagnostics/health.py @@ -133,15 +133,28 @@ def _table_date_summary(conn, table_name: str) -> Dict[str, Any]: except Exception as exc: return {"ok": False, "error": str(exc), "row_count": 0, "cities_count": 0} + max_date = row["max_date"] return { "ok": True, "row_count": int(row["row_count"] or 0), "cities_count": int(row["cities_count"] or 0), "min_date": row["min_date"], - "max_date": row["max_date"], + "max_date": max_date, + "stale_days": _target_date_stale_days(max_date), } +def _target_date_stale_days(target_date: Optional[str]) -> Optional[int]: + raw = str(target_date or "").strip() + if not raw: + return None + try: + parsed = datetime.strptime(raw, "%Y-%m-%d").date() + except Exception: + return None + return max(0, (datetime.now(timezone.utc).date() - parsed).days) + + def _truth_source_counts(conn) -> Dict[str, int]: try: rows = conn.execute( @@ -293,11 +306,17 @@ def _training_data_summary(account_db, city_registry) -> Dict[str, Any]: import sqlite3 db_path = account_db.db_path + daily_records = {"ok": False, "row_count": 0, "cities_count": 0} truth_records = {"ok": False, "row_count": 0, "cities_count": 0} truth_revisions = {"ok": False, "row_count": 0} training_features = {"ok": False, "row_count": 0, "cities_count": 0} + stale_threshold_days = max( + 0, + int(os.getenv("POLYWEATHER_TRAINING_DATA_STALE_WARN_DAYS", "2") or "2"), + ) try: with connect_sqlite(db_path, row_factory=sqlite3.Row) as conn: + daily_records = _table_date_summary(conn, "daily_records_store") truth_records = _table_date_summary(conn, "truth_records_store") if truth_records.get("ok"): truth_records["source_counts"] = _truth_source_counts(conn) @@ -311,19 +330,36 @@ def _training_data_summary(account_db, city_registry) -> Dict[str, Any]: "db_path": db_path, "db_ok": False, "error": str(exc), + "daily_records": daily_records, "truth_records": truth_records, "truth_revisions": truth_revisions, "training_features": training_features, + "stale": True, + "stale_threshold_days": stale_threshold_days, "city_coverage": {}, "model_city_coverage": {}, } + stale_days = [ + value + for value in ( + daily_records.get("stale_days"), + truth_records.get("stale_days"), + training_features.get("stale_days"), + ) + if value is not None + ] return { "db_path": db_path, "db_ok": True, + "daily_records": daily_records, "truth_records": truth_records, "truth_revisions": truth_revisions, "training_features": training_features, + "stale": any(int(value) > stale_threshold_days for value in stale_days) + if stale_days + else True, + "stale_threshold_days": stale_threshold_days, "city_coverage": city_coverage, "model_city_coverage": _model_city_coverage_summary( city_coverage.get("entries") or [], diff --git a/web/services/system_api.py b/web/services/system_api.py index eee60932..7c1bc887 100644 --- a/web/services/system_api.py +++ b/web/services/system_api.py @@ -5,6 +5,7 @@ from __future__ import annotations import time import os from collections import Counter +from datetime import datetime, timezone from typing import Any, Dict, Optional from fastapi import BackgroundTasks, Request @@ -13,6 +14,7 @@ from fastapi.responses import PlainTextResponse from loguru import logger from src.database.db_manager import DBManager +from src.database.sqlite_connection import connect_sqlite from src.utils.metrics import export_prometheus_metrics, gauge_set from web.core import build_health_payload, build_system_status_payload import web.routes as legacy_routes @@ -211,3 +213,51 @@ def _refresh_operational_metrics() -> None: gauge_set("polyweather_sqlite_db_size_bytes", os.path.getsize(db_path)) except OSError: pass + _refresh_training_data_metrics(db_path) + + +def _target_date_stale_days(target_date: Optional[str]) -> Optional[int]: + raw = str(target_date or "").strip() + if not raw: + return None + try: + parsed = datetime.strptime(raw, "%Y-%m-%d").date() + except Exception: + return None + return max(0, (datetime.now(timezone.utc).date() - parsed).days) + + +def _max_target_date(db_path: str, table_name: str) -> Optional[str]: + with connect_sqlite(db_path) as conn: + row = conn.execute(f"SELECT MAX(target_date) AS max_date FROM {table_name}").fetchone() + if not row: + return None + return row[0] + + +def _refresh_training_data_metrics(db_path: str) -> None: + stale_threshold_days = max( + 0, + int(os.getenv("POLYWEATHER_TRAINING_DATA_STALE_WARN_DAYS", "2") or "2"), + ) + table_metrics = { + "daily_records_store": "polyweather_daily_records_stale_days", + "truth_records_store": "polyweather_truth_records_stale_days", + "training_feature_records_store": "polyweather_training_features_stale_days", + } + max_stale_days: Optional[int] = None + for table_name, metric_name in table_metrics.items(): + try: + stale_days = _target_date_stale_days(_max_target_date(db_path, table_name)) + except Exception: + stale_days = None + if stale_days is None: + gauge_set(metric_name, -1) + max_stale_days = max(max_stale_days or 0, stale_threshold_days + 1) + continue + gauge_set(metric_name, stale_days) + max_stale_days = max(max_stale_days or 0, stale_days) + gauge_set( + "polyweather_training_data_stale", + 1 if max_stale_days is None or max_stale_days > stale_threshold_days else 0, + ) diff --git a/web/training_settlement_service.py b/web/training_settlement_service.py new file mode 100644 index 00000000..867d617e --- /dev/null +++ b/web/training_settlement_service.py @@ -0,0 +1,142 @@ +"""Low-frequency DEB training settlement maintenance.""" + +from __future__ import annotations + +from typing import Any, Callable, Dict, Iterable, Mapping, Optional, Sequence + +from loguru import logger + +from src.analysis.deb_algorithm import reconcile_recent_actual_highs +from src.data_collection.city_registry import CITY_REGISTRY + + +AnalysisRunner = Callable[[str], Mapping[str, Any]] +ActualReconciler = Callable[..., Mapping[str, Any]] + +UNSUPPORTED_SETTLEMENT_SOURCES = {"wunderground"} +RECONCILE_SETTLEMENT_SOURCES = {"metar", "hko", "noaa"} + + +def _normalize_city(city: str) -> str: + return str(city or "").strip().lower().replace("-", " ") + + +def _selected_city_names( + city_registry: Mapping[str, Mapping[str, Any]], + cities: Optional[Iterable[str]], +) -> Sequence[str]: + selected = {_normalize_city(city) for city in (cities or []) if _normalize_city(city)} + names = [] + for city in sorted(city_registry.keys()): + normalized = _normalize_city(city) + if selected and normalized not in selected: + continue + names.append(normalized) + return tuple(names) + + +def _is_supported_training_city(city_meta: Mapping[str, Any]) -> bool: + source = str(city_meta.get("settlement_source") or "metar").strip().lower() + if source in UNSUPPORTED_SETTLEMENT_SOURCES: + return False + return bool( + str(city_meta.get("icao") or "").strip() + or str(city_meta.get("settlement_station_code") or "").strip() + or source in {"hko", "cwa", "ims", "ncm", "aeroweb"} + ) + + +def _can_reconcile_actual_history(city_meta: Mapping[str, Any]) -> bool: + source = str(city_meta.get("settlement_source") or "metar").strip().lower() + return source in RECONCILE_SETTLEMENT_SOURCES + + +def _default_analysis_runner(city: str) -> Mapping[str, Any]: + from web.analysis_service import _analyze + + return _analyze(city, force_refresh=False, detail_mode="panel") + + +def _default_actual_reconciler(city: str, *, lookback_days: int) -> Mapping[str, Any]: + return reconcile_recent_actual_highs(city, lookback_days=lookback_days) + + +def run_training_settlement_cycle( + *, + city_registry: Optional[Mapping[str, Mapping[str, Any]]] = None, + cities: Optional[Iterable[str]] = None, + analysis_runner: Optional[AnalysisRunner] = None, + actual_reconciler: Optional[ActualReconciler] = None, + lookback_days: int = 10, +) -> Dict[str, Any]: + registry = city_registry or CITY_REGISTRY + run_analysis = analysis_runner or _default_analysis_runner + reconcile_actual = actual_reconciler or _default_actual_reconciler + safe_lookback = max(1, int(lookback_days or 1)) + + processed = 0 + failed = 0 + unsupported = 0 + items = [] + + for city in _selected_city_names(registry, cities): + meta = registry.get(city) or {} + if not _is_supported_training_city(meta): + unsupported += 1 + items.append( + { + "city": city, + "ok": True, + "status": "unsupported", + "reason": "unsupported_settlement_source", + } + ) + continue + + try: + analysis_payload = run_analysis(city) + if _can_reconcile_actual_history(meta): + reconcile_payload = reconcile_actual(city, lookback_days=safe_lookback) + else: + reconcile_payload = { + "ok": True, + "updated": 0, + "reason": "unsupported_reconcile_source", + "source": str(meta.get("settlement_source") or "").strip().lower(), + } + reconcile_ok = bool(reconcile_payload.get("ok", True)) + if reconcile_ok: + processed += 1 + else: + failed += 1 + items.append( + { + "city": city, + "ok": reconcile_ok, + "status": "processed" if reconcile_ok else "failed", + "analysis_status": str( + (analysis_payload or {}).get("status") or "ok" + ), + "reconcile": dict(reconcile_payload or {}), + } + ) + except Exception as exc: + failed += 1 + logger.warning("training settlement city failed city={}: {}", city, exc) + items.append( + { + "city": city, + "ok": False, + "status": "failed", + "error": str(exc), + } + ) + + return { + "ok": failed == 0, + "processed": processed, + "failed": failed, + "unsupported": unsupported, + "lookback_days": safe_lookback, + "items": items, + } diff --git a/web/training_settlement_worker.py b/web/training_settlement_worker.py new file mode 100644 index 00000000..7ac09247 --- /dev/null +++ b/web/training_settlement_worker.py @@ -0,0 +1,103 @@ +"""Standalone low-frequency DEB training settlement worker.""" + +from __future__ import annotations + +import argparse +import json +import os +import signal +import threading +import time +from types import FrameType +from typing import Optional + +from loguru import logger + +from web.training_settlement_service import run_training_settlement_cycle + + +_STOP_EVENT = threading.Event() + + +def _env_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + return int(raw) + except Exception: + return default + + +def _handle_stop_signal(signum: int, _frame: Optional[FrameType]) -> None: + logger.info("training settlement worker stopping signal={}", signum) + _STOP_EVENT.set() + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Run low-frequency DEB training settlement maintenance." + ) + parser.add_argument("--once", action="store_true", help="Run one cycle and exit.") + parser.add_argument( + "--interval-sec", + type=int, + default=_env_int("POLYWEATHER_TRAINING_SETTLEMENT_INTERVAL_SEC", 21600), + ) + parser.add_argument( + "--initial-delay-sec", + type=int, + default=_env_int("POLYWEATHER_TRAINING_SETTLEMENT_INITIAL_DELAY_SEC", 60), + ) + parser.add_argument( + "--lookback-days", + type=int, + default=_env_int("POLYWEATHER_TRAINING_SETTLEMENT_LOOKBACK_DAYS", 10), + ) + parser.add_argument("--cities", nargs="*", default=None) + return parser.parse_args() + + +def _run_once(*, lookback_days: int, cities: Optional[list[str]]) -> dict: + result = run_training_settlement_cycle( + cities=cities, + lookback_days=lookback_days, + ) + logger.info("training settlement result={}", json.dumps(result, ensure_ascii=False)) + return result + + +def main() -> None: + signal.signal(signal.SIGINT, _handle_stop_signal) + signal.signal(signal.SIGTERM, _handle_stop_signal) + args = _parse_args() + + if args.once: + result = _run_once(lookback_days=args.lookback_days, cities=args.cities) + print(json.dumps(result, ensure_ascii=False)) + return + + interval_sec = max(300, int(args.interval_sec or 21600)) + initial_delay_sec = max(0, int(args.initial_delay_sec or 0)) + logger.info( + "training settlement worker started interval={}s lookback_days={}", + interval_sec, + args.lookback_days, + ) + if initial_delay_sec and _STOP_EVENT.wait(initial_delay_sec): + return + + while not _STOP_EVENT.is_set(): + started = time.time() + try: + _run_once(lookback_days=args.lookback_days, cities=args.cities) + except Exception as exc: + logger.exception("training settlement cycle failed: {}", exc) + elapsed = time.time() - started + wait_for = max(5.0, interval_sec - elapsed) + if _STOP_EVENT.wait(wait_for): + break + + +if __name__ == "__main__": + main()