Add dashboard prewarm worker and cache visibility

web/analysis_service.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import re
 import time as _time
+import threading
 from datetime import datetime, timezone, timedelta
 from typing import Dict, Any, Optional
 
@@ -27,6 +28,44 @@ from src.data_collection.city_registry import ALIASES
 from src.models.lgbm_daily_high import predict_lgbm_daily_high
 
 TURKISH_MGM_CITIES = {"ankara", "istanbul"}
+_ANALYSIS_CACHE_STATS_LOCK = threading.Lock()
+_ANALYSIS_CACHE_STATS: Dict[str, Any] = {
+    "total_requests": 0,
+    "cache_hits": 0,
+    "cache_misses": 0,
+    "force_refresh_requests": 0,
+    "last_cache_hit_at": None,
+    "last_cache_miss_at": None,
+    "last_city": None,
+}
+
+
+def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) -> None:
+    now = datetime.now(timezone.utc).isoformat()
+    with _ANALYSIS_CACHE_STATS_LOCK:
+        _ANALYSIS_CACHE_STATS["total_requests"] = int(_ANALYSIS_CACHE_STATS.get("total_requests") or 0) + 1
+        _ANALYSIS_CACHE_STATS["last_city"] = str(city or "")
+        if force_refresh:
+            _ANALYSIS_CACHE_STATS["force_refresh_requests"] = int(_ANALYSIS_CACHE_STATS.get("force_refresh_requests") or 0) + 1
+        if hit:
+            _ANALYSIS_CACHE_STATS["cache_hits"] = int(_ANALYSIS_CACHE_STATS.get("cache_hits") or 0) + 1
+            _ANALYSIS_CACHE_STATS["last_cache_hit_at"] = now
+        else:
+            _ANALYSIS_CACHE_STATS["cache_misses"] = int(_ANALYSIS_CACHE_STATS.get("cache_misses") or 0) + 1
+            _ANALYSIS_CACHE_STATS["last_cache_miss_at"] = now
+
+
+def get_analysis_cache_stats() -> Dict[str, Any]:
+    with _ANALYSIS_CACHE_STATS_LOCK:
+        stats = dict(_ANALYSIS_CACHE_STATS)
+    hits = int(stats.get("cache_hits") or 0)
+    misses = int(stats.get("cache_misses") or 0)
+    eligible = hits + misses
+    hit_rate = (hits / eligible) if eligible > 0 else None
+    miss_rate = (misses / eligible) if eligible > 0 else None
+    stats["hit_rate"] = round(hit_rate, 4) if hit_rate is not None else None
+    stats["miss_rate"] = round(miss_rate, 4) if miss_rate is not None else None
+    return stats
 
 
 def _interpolate_hourly_value(
@@ -780,7 +819,9 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]:
     if not force_refresh:
         cached = _cache.get(city)
         if cached and _time.time() - cached["t"] < ttl:
+            _record_analysis_cache_event(city=city, hit=True, force_refresh=False)
             return cached["d"]
+    _record_analysis_cache_event(city=city, hit=False, force_refresh=force_refresh)
 
     info = CITIES[city]
     lat, lon, is_f = info["lat"], info["lon"], info["f"]

web/core.py

@@ -21,6 +21,7 @@ from src.data_collection.country_networks import provider_coverage_summary
 from src.data_collection.city_risk_profiles import CITY_RISK_PROFILES  # noqa: F401
 from src.data_collection.polymarket_readonly import PolymarketReadOnlyLayer
 from src.auth.supabase_entitlement import SUPABASE_ENTITLEMENT, extract_bearer_token
+from src.utils.prewarm_dashboard import get_prewarm_runtime_summary
 from src.utils.metrics import (
     build_metrics_summary,
     counter_inc,
@@ -391,24 +392,34 @@ def _sqlite_health() -> Dict[str, Any]:
 
 
 def _cache_summary() -> Dict[str, Any]:
+    from web.analysis_service import get_analysis_cache_stats
+
+    open_meteo_forecast_entries = len(getattr(_weather, "_open_meteo_cache", {}) or {})
+    open_meteo_ensemble_entries = len(getattr(_weather, "_ensemble_cache", {}) or {})
+    open_meteo_multi_model_entries = len(getattr(_weather, "_multi_model_cache", {}) or {})
+    metar_entries = len(getattr(_weather, "_metar_cache", {}) or {})
+    taf_entries = len(getattr(_weather, "_taf_cache", {}) or {})
+    nmc_entries = len(getattr(_weather, "_nmc_cache", {}) or {})
+    settlement_entries = len(getattr(_weather, "_settlement_cache", {}) or {})
+
     gauge_set("polyweather_api_cache_entries", len(_cache))
-    gauge_set(
-        "polyweather_open_meteo_forecast_cache_entries",
-        len(getattr(_weather, "_open_meteo_cache", {}) or {}),
-    )
-    gauge_set(
-        "polyweather_open_meteo_ensemble_cache_entries",
-        len(getattr(_weather, "_ensemble_cache", {}) or {}),
-    )
-    gauge_set(
-        "polyweather_open_meteo_multi_model_cache_entries",
-        len(getattr(_weather, "_multi_model_cache", {}) or {}),
-    )
+    gauge_set("polyweather_open_meteo_forecast_cache_entries", open_meteo_forecast_entries)
+    gauge_set("polyweather_open_meteo_ensemble_cache_entries", open_meteo_ensemble_entries)
+    gauge_set("polyweather_open_meteo_multi_model_cache_entries", open_meteo_multi_model_entries)
+    gauge_set("polyweather_metar_cache_entries", metar_entries)
+    gauge_set("polyweather_taf_cache_entries", taf_entries)
+    gauge_set("polyweather_nmc_cache_entries", nmc_entries)
+    gauge_set("polyweather_settlement_cache_entries", settlement_entries)
     return {
         "api_cache_entries": len(_cache),
-        "open_meteo_forecast_entries": len(getattr(_weather, "_open_meteo_cache", {}) or {}),
-        "open_meteo_ensemble_entries": len(getattr(_weather, "_ensemble_cache", {}) or {}),
-        "open_meteo_multi_model_entries": len(getattr(_weather, "_multi_model_cache", {}) or {}),
+        "open_meteo_forecast_entries": open_meteo_forecast_entries,
+        "open_meteo_ensemble_entries": open_meteo_ensemble_entries,
+        "open_meteo_multi_model_entries": open_meteo_multi_model_entries,
+        "metar_entries": metar_entries,
+        "taf_entries": taf_entries,
+        "nmc_entries": nmc_entries,
+        "settlement_entries": settlement_entries,
+        "analysis": get_analysis_cache_stats(),
     }
 
 
@@ -750,5 +761,6 @@ def build_system_status_payload() -> Dict[str, Any]:
         "probability": _probability_summary(),
         "training_data": _training_data_summary(),
         "station_networks": provider_coverage_summary(),
+        "prewarm": get_prewarm_runtime_summary(),
         "cities_count": len(CITIES),
     }

web/routes.py

@@ -1,9 +1,11 @@
 from __future__ import annotations
 
 import os
+import time
 from datetime import datetime, timedelta
 from typing import Optional
 
+from fastapi.concurrency import run_in_threadpool
 from fastapi import APIRouter, HTTPException, Request
 from fastapi.responses import PlainTextResponse
 from loguru import logger
@@ -66,6 +68,23 @@ TRACKABLE_ANALYTICS_EVENTS = {
     "checkout_succeeded",
 }
 
+DEFAULT_PREWARM_CITIES = [
+    "ankara",
+    "istanbul",
+    "shanghai",
+    "beijing",
+    "shenzhen",
+    "wuhan",
+    "chengdu",
+    "chongqing",
+    "hong kong",
+    "taipei",
+    "london",
+    "paris",
+    "new york",
+    "los angeles",
+]
+
 
 def _parse_snapshot_dt(value: object) -> Optional[datetime]:
     raw = str(value or "").strip()
@@ -178,6 +197,20 @@ def _normalize_city_or_404(name: str) -> str:
     return city
 
 
+def _normalize_city_list(raw: Optional[str]) -> list[str]:
+    if not raw:
+        return list(DEFAULT_PREWARM_CITIES)
+    out: list[str] = []
+    for part in str(raw).split(","):
+        city = str(part or "").strip().lower().replace("-", " ")
+        if not city:
+            continue
+        city = ALIASES.get(city, city)
+        if city in CITIES and city not in out:
+            out.append(city)
+    return out
+
+
 def _history_file_path() -> str:
     project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     return os.path.join(project_root, "data", "daily_records.json")
@@ -251,7 +284,85 @@ async def healthz():
 
 @router.get("/api/system/status")
 async def system_status():
-    return build_system_status_payload()
+    return await run_in_threadpool(build_system_status_payload)
+
+
+@router.post("/api/system/prewarm")
+async def system_prewarm(
+    request: Request,
+    cities: Optional[str] = None,
+    force_refresh: bool = False,
+    include_detail: bool = False,
+    include_market: bool = False,
+):
+    _assert_entitlement(request)
+    selected = _normalize_city_list(cities)
+    if not selected:
+        raise HTTPException(status_code=400, detail="No valid cities to prewarm")
+
+    started = time.perf_counter()
+    warmed: list[dict[str, object]] = []
+    failed: list[dict[str, object]] = []
+    summary_ok = 0
+    detail_ok = 0
+    market_ok = 0
+
+    for city in selected:
+        city_started = time.perf_counter()
+        try:
+            data = _analyze(city, force_refresh=force_refresh)
+            entry: dict[str, object] = {
+                "city": city,
+                "summary": True,
+                "duration_ms": round((time.perf_counter() - city_started) * 1000.0, 1),
+            }
+            summary_ok += 1
+            if include_detail:
+                _build_city_summary_payload(data)
+                _build_city_detail_payload(
+                    data,
+                    target_date=str(data.get("local_date") or "").strip() or None,
+                )
+                entry["detail"] = True
+                detail_ok += 1
+            if include_market:
+                _build_city_detail_payload(
+                    data,
+                    target_date=str(data.get("local_date") or "").strip() or None,
+                )
+                entry["market"] = True
+                market_ok += 1
+            warmed.append(entry)
+        except Exception as exc:
+            failed.append(
+                {
+                    "city": city,
+                    "error": str(exc),
+                    "duration_ms": round((time.perf_counter() - city_started) * 1000.0, 1),
+                }
+            )
+
+    total_ms = round((time.perf_counter() - started) * 1000.0, 1)
+    logger.info(
+        "system prewarm finished count={} failed={} force_refresh={} include_detail={} include_market={} duration_ms={}",
+        len(warmed),
+        len(failed),
+        force_refresh,
+        include_detail,
+        include_market,
+        total_ms,
+    )
+    return {
+        "ok": len(failed) == 0,
+        "cities": selected,
+        "warmed": warmed,
+        "failed": failed,
+        "summary_ok": summary_ok,
+        "detail_ok": detail_ok,
+        "market_ok": market_ok,
+        "failed_count": len(failed),
+        "duration_ms": total_ms,
+    }
 
 
 @router.get("/metrics", response_class=PlainTextResponse)
@@ -264,7 +375,7 @@ async def metrics():
 
 @router.get("/api/cities")
 async def list_cities(request: Request):
-    try:
+    def _build_payload():
         out = []
         deb_recent_index = _build_recent_deb_performance_index()
         for name, info in CITIES.items():
@@ -302,6 +413,9 @@ async def list_cities(request: Request):
                 }
             )
         return {"cities": out}
+
+    try:
+        return await run_in_threadpool(_build_payload)
     except Exception as exc:
         logger.error(f"Error in list_cities: {exc}")
         raise HTTPException(status_code=500, detail=str(exc)) from exc
@@ -310,105 +424,110 @@ async def list_cities(request: Request):
 @router.get("/api/city/{name}")
 async def city_detail(request: Request, name: str, force_refresh: bool = False):
     _assert_entitlement(request)
-    return _analyze(_normalize_city_or_404(name), force_refresh=force_refresh)
+    city = _normalize_city_or_404(name)
+    return await run_in_threadpool(_analyze, city, force_refresh)
 
 
 @router.get("/api/history/{name}")
 async def city_history(request: Request, name: str):
     _assert_entitlement(request)
     city = _normalize_city_or_404(name)
-    source = str(CITIES.get(city, {}).get("settlement_source") or "metar").strip().lower()
-    truth_rows = _truth_record_repo.load_city(city)
-    feature_rows = _training_feature_repo.load_city(city)
 
-    if not truth_rows and not feature_rows:
-        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-        history_file = os.path.join(project_root, "data", "daily_records.json")
-        data = load_history(history_file)
-        city_data = data.get(city, {}) if isinstance(data.get(city, {}), dict) else {}
-    else:
-        all_dates = sorted(set(truth_rows.keys()) | set(feature_rows.keys()))
-        city_data = {}
-        for day in all_dates:
-            record: dict[str, object] = {}
-            truth = truth_rows.get(day) or {}
-            features = feature_rows.get(day) or {}
-            if truth.get("actual_high") is not None:
-                record["actual_high"] = truth.get("actual_high")
-                record["settlement_source"] = truth.get("settlement_source")
-                record["settlement_station_code"] = truth.get("settlement_station_code")
-                record["settlement_station_label"] = truth.get("settlement_station_label")
-                record["truth_version"] = truth.get("truth_version")
-                record["updated_by"] = truth.get("updated_by")
-                record["truth_updated_at"] = truth.get("truth_updated_at")
-            if isinstance(features, dict):
-                if features.get("deb_prediction") is not None:
-                    record["deb_prediction"] = features.get("deb_prediction")
-                if features.get("mu") is not None:
-                    record["mu"] = features.get("mu")
-                if isinstance(features.get("forecasts"), dict):
-                    record["forecasts"] = features.get("forecasts")
-            city_data[day] = record
+    def _build_history_payload():
+        source = str(CITIES.get(city, {}).get("settlement_source") or "metar").strip().lower()
+        truth_rows = _truth_record_repo.load_city(city)
+        feature_rows = _training_feature_repo.load_city(city)
+
+        if not truth_rows and not feature_rows:
+            project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+            history_file = os.path.join(project_root, "data", "daily_records.json")
+            data = load_history(history_file)
+            city_data = data.get(city, {}) if isinstance(data.get(city, {}), dict) else {}
+        else:
+            all_dates = sorted(set(truth_rows.keys()) | set(feature_rows.keys()))
+            city_data = {}
+            for day in all_dates:
+                record: dict[str, object] = {}
+                truth = truth_rows.get(day) or {}
+                features = feature_rows.get(day) or {}
+                if truth.get("actual_high") is not None:
+                    record["actual_high"] = truth.get("actual_high")
+                    record["settlement_source"] = truth.get("settlement_source")
+                    record["settlement_station_code"] = truth.get("settlement_station_code")
+                    record["settlement_station_label"] = truth.get("settlement_station_label")
+                    record["truth_version"] = truth.get("truth_version")
+                    record["updated_by"] = truth.get("updated_by")
+                    record["truth_updated_at"] = truth.get("truth_updated_at")
+                if isinstance(features, dict):
+                    if features.get("deb_prediction") is not None:
+                        record["deb_prediction"] = features.get("deb_prediction")
+                    if features.get("mu") is not None:
+                        record["mu"] = features.get("mu")
+                    if isinstance(features.get("forecasts"), dict):
+                        record["forecasts"] = features.get("forecasts")
+                city_data[day] = record
+
+        if not city_data:
+            return {
+                "history": [],
+                "settlement_source": source,
+                "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()),
+            }
+
+        out = []
+        for day, rec in sorted(city_data.items()):
+            if not isinstance(rec, dict):
+                rec = {}
+
+            act = rec.get("actual_high")
+            deb = rec.get("deb_prediction")
+            mu = rec.get("mu")
+            snapshots = load_snapshot_rows_for_day(city, day)
+            peak_ref = _build_peak_minus_12h_reference(
+                actual_high=act,
+                snapshots=snapshots,
+            )
+            forecasts_raw = rec.get("forecasts", {}) or {}
+            forecasts = {}
+            if isinstance(forecasts_raw, dict):
+                for model_name, model_value in forecasts_raw.items():
+                    if _is_excluded_model_name(str(model_name)):
+                        continue
+                    fv = _sf(model_value)
+                    forecasts[str(model_name)] = fv if fv is not None else None
+            forecasts = _merge_missing_history_forecasts_from_snapshots(
+                forecasts,
+                snapshots,
+            )
+            mgm = forecasts.get("MGM")
+            out.append(
+                {
+                    "date": day,
+                    "actual": float(act) if act is not None else None,
+                    "deb": float(deb) if deb is not None else None,
+                    "mu": float(mu) if mu is not None else None,
+                    "mgm": float(mgm) if mgm is not None else None,
+                    "forecasts": forecasts,
+                    "settlement_source": rec.get("settlement_source"),
+                    "settlement_station_code": rec.get("settlement_station_code"),
+                    "settlement_station_label": rec.get("settlement_station_label"),
+                    "truth_version": rec.get("truth_version"),
+                    "updated_by": rec.get("updated_by"),
+                    "truth_updated_at": rec.get("truth_updated_at"),
+                    "actual_peak_time": peak_ref.get("actual_peak_time"),
+                    "deb_at_peak_minus_12h": peak_ref.get("deb_at_peak_minus_12h"),
+                    "deb_at_peak_minus_12h_time": peak_ref.get("deb_at_peak_minus_12h_time"),
+                    "deb_at_peak_minus_12h_error": peak_ref.get("deb_at_peak_minus_12h_error"),
+                }
+            )
 
-    if not city_data:
         return {
-            "history": [],
+            "history": out,
             "settlement_source": source,
             "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()),
         }
 
-    out = []
-    for day, rec in sorted(city_data.items()):
-        if not isinstance(rec, dict):
-            rec = {}
-
-        act = rec.get("actual_high")
-        deb = rec.get("deb_prediction")
-        mu = rec.get("mu")
-        snapshots = load_snapshot_rows_for_day(city, day)
-        peak_ref = _build_peak_minus_12h_reference(
-            actual_high=act,
-            snapshots=snapshots,
-        )
-        forecasts_raw = rec.get("forecasts", {}) or {}
-        forecasts = {}
-        if isinstance(forecasts_raw, dict):
-            for model_name, model_value in forecasts_raw.items():
-                if _is_excluded_model_name(str(model_name)):
-                    continue
-                fv = _sf(model_value)
-                forecasts[str(model_name)] = fv if fv is not None else None
-        forecasts = _merge_missing_history_forecasts_from_snapshots(
-            forecasts,
-            snapshots,
-        )
-        mgm = forecasts.get("MGM")
-        out.append(
-            {
-                "date": day,
-                "actual": float(act) if act is not None else None,
-                "deb": float(deb) if deb is not None else None,
-                "mu": float(mu) if mu is not None else None,
-                "mgm": float(mgm) if mgm is not None else None,
-                "forecasts": forecasts,
-                "settlement_source": rec.get("settlement_source"),
-                "settlement_station_code": rec.get("settlement_station_code"),
-                "settlement_station_label": rec.get("settlement_station_label"),
-                "truth_version": rec.get("truth_version"),
-                "updated_by": rec.get("updated_by"),
-                "truth_updated_at": rec.get("truth_updated_at"),
-                "actual_peak_time": peak_ref.get("actual_peak_time"),
-                "deb_at_peak_minus_12h": peak_ref.get("deb_at_peak_minus_12h"),
-                "deb_at_peak_minus_12h_time": peak_ref.get("deb_at_peak_minus_12h_time"),
-                "deb_at_peak_minus_12h_error": peak_ref.get("deb_at_peak_minus_12h_error"),
-            }
-        )
-
-    return {
-        "history": out,
-        "settlement_source": source,
-        "settlement_source_label": SETTLEMENT_SOURCE_LABELS.get(source, source.upper()),
-    }
+    return await run_in_threadpool(_build_history_payload)
 
 
 @router.get("/api/auth/me")
@@ -898,8 +1017,9 @@ async def payment_reconcile_latest(request: Request):
 
 @router.get("/api/city/{name}/summary")
 async def city_summary(request: Request, name: str, force_refresh: bool = False):
-    data = _analyze(_normalize_city_or_404(name), force_refresh=force_refresh)
-    return _build_city_summary_payload(data)
+    city = _normalize_city_or_404(name)
+    data = await run_in_threadpool(_analyze, city, force_refresh)
+    return await run_in_threadpool(_build_city_summary_payload, data)
 
 
 @router.get("/api/city/{name}/detail")
@@ -911,9 +1031,11 @@ async def city_detail_aggregate(
     target_date: Optional[str] = None,
 ):
     _assert_entitlement(request)
-    data = _analyze(_normalize_city_or_404(name), force_refresh=force_refresh)
-    return _build_city_detail_payload(
+    city = _normalize_city_or_404(name)
+    data = await run_in_threadpool(_analyze, city, force_refresh)
+    return await run_in_threadpool(
+        _build_city_detail_payload,
         data,
-        market_slug=market_slug,
-        target_date=target_date,
+        market_slug,
+        target_date,
     )