From c3da29c09cec3552040d43e4ef8bc65bd3a2f12a Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 8 Apr 2026 06:53:49 +0800 Subject: [PATCH] Add dashboard prewarm worker and cache visibility --- docker-compose.yml | 11 + frontend/app/loading.tsx | 5 + .../components/dashboard/DashboardEntry.tsx | 2 + .../dashboard/DashboardShellSkeleton.tsx | 105 ++++++ frontend/components/ops/OpsDashboard.tsx | 128 ++++++++ requirements.txt | 1 + scripts/prewarm_dashboard_cache.py | 49 +++ scripts/prewarm_dashboard_worker.py | 65 ++++ src/bot/runtime_coordinator.py | 27 ++ src/data_collection/metar_sources.py | 6 +- src/data_collection/mgm_sources.py | 14 +- src/data_collection/nmc_sources.py | 20 +- src/data_collection/nws_open_meteo_sources.py | 14 +- src/data_collection/polymarket_readonly.py | 7 +- src/data_collection/settlement_sources.py | 20 +- src/data_collection/weather_sources.py | 91 ++++- src/utils/prewarm_dashboard.py | 305 +++++++++++++++++ tests/test_web_observability.py | 6 + web/analysis_service.py | 41 +++ web/core.py | 42 ++- web/routes.py | 310 ++++++++++++------ 21 files changed, 1120 insertions(+), 149 deletions(-) create mode 100644 frontend/app/loading.tsx create mode 100644 frontend/components/dashboard/DashboardShellSkeleton.tsx create mode 100644 scripts/prewarm_dashboard_cache.py create mode 100644 scripts/prewarm_dashboard_worker.py create mode 100644 src/utils/prewarm_dashboard.py diff --git a/docker-compose.yml b/docker-compose.yml index 8a5b7741..083c5ab6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -34,6 +34,17 @@ services: # UID/GID are mainly useful on Linux hosts to avoid root-owned output files. user: "${UID:-1000}:${GID:-1000}" + polyweather_prewarm: + <<: *polyweather-base + container_name: polyweather_prewarm + restart: unless-stopped + profiles: ["workers"] + command: python scripts/prewarm_dashboard_worker.py --include-detail --include-market + volumes: + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather + - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data + user: "${UID:-1000}:${GID:-1000}" + polyweather_prometheus: image: prom/prometheus:v3.4.1 container_name: polyweather_prometheus diff --git a/frontend/app/loading.tsx b/frontend/app/loading.tsx new file mode 100644 index 00000000..e99baf0e --- /dev/null +++ b/frontend/app/loading.tsx @@ -0,0 +1,5 @@ +import { DashboardShellSkeleton } from "@/components/dashboard/DashboardShellSkeleton"; + +export default function Loading() { + return ; +} diff --git a/frontend/components/dashboard/DashboardEntry.tsx b/frontend/components/dashboard/DashboardEntry.tsx index e4cfaec6..ee914549 100644 --- a/frontend/components/dashboard/DashboardEntry.tsx +++ b/frontend/components/dashboard/DashboardEntry.tsx @@ -1,6 +1,7 @@ "use client"; import dynamic from "next/dynamic"; +import { DashboardShellSkeleton } from "@/components/dashboard/DashboardShellSkeleton"; const PolyWeatherDashboard = dynamic( () => @@ -9,6 +10,7 @@ const PolyWeatherDashboard = dynamic( ), { ssr: false, + loading: () => , }, ); diff --git a/frontend/components/dashboard/DashboardShellSkeleton.tsx b/frontend/components/dashboard/DashboardShellSkeleton.tsx new file mode 100644 index 00000000..f1b78294 --- /dev/null +++ b/frontend/components/dashboard/DashboardShellSkeleton.tsx @@ -0,0 +1,105 @@ +"use client"; + +import { Skeleton } from "@/components/ui/skeleton"; + +export function DashboardShellSkeleton() { + return ( +
+
+
+ + +
+
+ + +
+
+ +
+
+ + {Array.from({ length: 6 }).map((_, index) => ( + + ))} +
+ +
+ + {Array.from({ length: 8 }).map((_, index) => ( + + ))} +
+ +
+ + + +
+
+
+ ); +} diff --git a/frontend/components/ops/OpsDashboard.tsx b/frontend/components/ops/OpsDashboard.tsx index 8b35fb20..30d23432 100644 --- a/frontend/components/ops/OpsDashboard.tsx +++ b/frontend/components/ops/OpsDashboard.tsx @@ -23,6 +23,27 @@ type SystemStatusPayload = { db?: { ok?: boolean; db_path?: string }; features?: Record; metrics?: Record; + cache?: { + api_cache_entries?: number; + open_meteo_forecast_entries?: number; + open_meteo_ensemble_entries?: number; + open_meteo_multi_model_entries?: number; + metar_entries?: number; + taf_entries?: number; + nmc_entries?: number; + settlement_entries?: number; + analysis?: { + total_requests?: number; + cache_hits?: number; + cache_misses?: number; + force_refresh_requests?: number; + last_cache_hit_at?: string | null; + last_cache_miss_at?: string | null; + last_city?: string | null; + hit_rate?: number | null; + miss_rate?: number | null; + }; + }; probability?: { engine_mode?: string; rollout?: { @@ -133,6 +154,39 @@ type SystemStatusPayload = { } >; }; + prewarm?: { + enabled?: boolean; + base_url?: string; + configured_cities?: string[]; + configured_city_count?: number; + interval_sec?: number; + jitter_sec?: number; + include_detail?: boolean; + include_market?: boolean; + force_refresh?: boolean; + thread_alive?: boolean; + runtime?: { + cycle_count?: number; + success_count?: number; + failure_count?: number; + last_started_at?: string | null; + last_finished_at?: string | null; + last_duration_sec?: number | null; + last_success?: boolean | null; + last_http_status?: number | null; + last_error?: string | null; + last_requested_cities?: string[]; + last_requested_city_count?: number; + last_include_detail?: boolean; + last_include_market?: boolean; + last_force_refresh?: boolean; + last_warmed_count?: number; + last_summary_ok?: number; + last_detail_ok?: number; + last_market_ok?: number; + last_failed_count?: number; + }; + }; }; type PaymentRuntimePayload = { @@ -433,6 +487,10 @@ export function OpsDashboard() { const modelCityCoverage = trainingData?.model_city_coverage; const trainingArtifacts = trainingData?.artifacts; const stationNetworks = status?.station_networks; + const cacheSummary = status?.cache; + const analysisCache = cacheSummary?.analysis; + const prewarm = status?.prewarm; + const prewarmRuntime = prewarm?.runtime; const truthSources = Object.entries(truthRecords?.source_counts || {}); const providerRows = useMemo( () => @@ -585,6 +643,7 @@ export function OpsDashboard() { const opsNav = [ { id: "overview", label: "总览" }, { id: "station-networks", label: "站网覆盖" }, + { id: "prewarm", label: "预热缓存" }, { id: "funnel", label: "转化漏斗" }, { id: "probability", label: "EMOS 门禁" }, { id: "training-data", label: "训练数据" }, @@ -746,6 +805,75 @@ export function OpsDashboard() { +
+ + + 预热 Worker + 确认后台定向预热是否真的在跑，以及最近一次有没有成功。 + + +
+ + + + +
+
+
最近开始{formatDateTime(prewarmRuntime?.last_started_at)}
+
最近结束{formatDateTime(prewarmRuntime?.last_finished_at)}
+
最近耗时{prewarmRuntime?.last_duration_sec ?? "-"} s
+
最近结果{prewarmRuntime?.last_success === true ? "success" : prewarmRuntime?.last_success === false ? "failed" : "-"}
+
最近 HTTP{prewarmRuntime?.last_http_status ?? "-"}
+
最近预热城市{prewarmRuntime?.last_requested_city_count ?? 0}
+
最近命中结果{prewarmRuntime?.last_warmed_count ?? 0}
+
summary / detail / market{prewarmRuntime?.last_summary_ok ?? 0} / {prewarmRuntime?.last_detail_ok ?? 0} / {prewarmRuntime?.last_market_ok ?? 0}
+
最近失败数{prewarmRuntime?.last_failed_count ?? 0}
+
累计 cycle{prewarmRuntime?.cycle_count ?? 0}
+
success / failed{prewarmRuntime?.success_count ?? 0} / {prewarmRuntime?.failure_count ?? 0}
+
+ {prewarmRuntime?.last_error ? ( +
+ 最近错误：{prewarmRuntime.last_error} +
+ ) : null} +
+ 当前预热目标：{(prewarmRuntime?.last_requested_cities || prewarm?.configured_cities || []).slice(0, 12).join(", ") || "未配置"} +
+ + + + + + 缓存桶状态 + 观察热点数据是不是已经被打热，避免每次请求都现场抓取。 + + +
+ + + + + + + + +
+
+
summary 请求总数{analysisCache?.total_requests ?? 0}
+
cache hit / miss{analysisCache?.cache_hits ?? 0} / {analysisCache?.cache_misses ?? 0}
+
hit rate{typeof analysisCache?.hit_rate === "number" ? `${(analysisCache.hit_rate * 100).toFixed(1)}%` : "-"}
+
force refresh{analysisCache?.force_refresh_requests ?? 0}
+
最近 hit{formatDateTime(analysisCache?.last_cache_hit_at)}
+
最近 miss{formatDateTime(analysisCache?.last_cache_miss_at)}
+
最近城市{analysisCache?.last_city || "-"}
+
+
+ 这里同时展示桶内条目数和 summary 层 cache hit/miss。detail / market 目前仍以最近一次预热结果为准。 +
+ + +
+ {error ? ( diff --git a/requirements.txt b/requirements.txt index ca96ec99..501dab9a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ requests +httpx loguru pyTelegramBotAPI python-dotenv diff --git a/scripts/prewarm_dashboard_cache.py b/scripts/prewarm_dashboard_cache.py new file mode 100644 index 00000000..e540bf65 --- /dev/null +++ b/scripts/prewarm_dashboard_cache.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def parse_args() -> argparse.Namespace: + from src.utils.prewarm_dashboard import DEFAULT_CITIES + + parser = argparse.ArgumentParser( + description="Prewarm PolyWeather summary/detail/market caches for selected cities.", + ) + parser.add_argument( + "--base-url", + default=os.getenv("POLYWEATHER_BACKEND_URL", "http://127.0.0.1:8000"), + help="Backend base URL, defaults to POLYWEATHER_BACKEND_URL or http://127.0.0.1:8000", + ) + parser.add_argument( + "--cities", + default=",".join(DEFAULT_CITIES), + help="Comma-separated city names to prewarm", + ) + parser.add_argument("--force-refresh", action="store_true") + parser.add_argument("--include-detail", action="store_true") + parser.add_argument("--include-market", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + from src.utils.prewarm_dashboard import run_prewarm + + return run_prewarm( + base_url=args.base_url, + cities=args.cities, + force_refresh=bool(args.force_refresh), + include_detail=bool(args.include_detail), + include_market=bool(args.include_market), + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/prewarm_dashboard_worker.py b/scripts/prewarm_dashboard_worker.py new file mode 100644 index 00000000..25019cfc --- /dev/null +++ b/scripts/prewarm_dashboard_worker.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import argparse +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def parse_args() -> argparse.Namespace: + from src.utils.prewarm_dashboard import DEFAULT_CITIES + + parser = argparse.ArgumentParser( + description="Run a background dashboard prewarm worker for hot PolyWeather cities.", + ) + parser.add_argument( + "--base-url", + default=os.getenv("POLYWEATHER_BACKEND_URL", "http://127.0.0.1:8000"), + help="Backend base URL, defaults to POLYWEATHER_BACKEND_URL or http://127.0.0.1:8000", + ) + parser.add_argument( + "--cities", + default=",".join(DEFAULT_CITIES), + help="Comma-separated city names to prewarm", + ) + parser.add_argument( + "--interval-sec", + type=int, + default=int(os.getenv("POLYWEATHER_PREWARM_INTERVAL_SEC", "300")), + help="Worker interval in seconds", + ) + parser.add_argument( + "--jitter-sec", + type=int, + default=int(os.getenv("POLYWEATHER_PREWARM_JITTER_SEC", "20")), + help="Random jitter added to each loop in seconds", + ) + parser.add_argument("--force-refresh", action="store_true") + parser.add_argument("--include-detail", action="store_true") + parser.add_argument("--include-market", action="store_true") + parser.add_argument("--once", action="store_true") + return parser.parse_args() + + +def main() -> int: + args = parse_args() + from src.utils.prewarm_dashboard import run_worker_loop + + return run_worker_loop( + base_url=args.base_url, + cities=args.cities, + interval_sec=args.interval_sec, + jitter_sec=args.jitter_sec, + force_refresh=bool(args.force_refresh), + include_detail=bool(args.include_detail), + include_market=bool(args.include_market), + once=bool(args.once), + ) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 0abd9bdc..4a8c2642 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -84,6 +84,7 @@ class StartupCoordinator: def start_all(self) -> RuntimeStatus: loops = [ self._start_trade_alert_loop(), + self._start_dashboard_prewarm_loop(), self._start_polygon_wallet_loop(), self._start_polymarket_wallet_activity_loop(), self._start_weekly_reward_loop(), @@ -179,6 +180,32 @@ class StartupCoordinator: ), ) + def _start_dashboard_prewarm_loop(self) -> LoopStatus: + enabled = _env_bool("POLYWEATHER_DASHBOARD_PREWARM_ENABLED", False) + interval = max(30, _env_int("POLYWEATHER_PREWARM_INTERVAL_SEC", 300)) + jitter = max(0, _env_int("POLYWEATHER_PREWARM_JITTER_SEC", 20)) + cities_count = _parse_csv_count(os.getenv("POLYWEATHER_PREWARM_CITIES")) or 14 + details = { + "interval_sec": interval, + "jitter_sec": jitter, + "cities_count": cities_count, + "include_detail": _env_bool("POLYWEATHER_PREWARM_INCLUDE_DETAIL", True), + "include_market": _env_bool("POLYWEATHER_PREWARM_INCLUDE_MARKET", True), + "force_refresh": _env_bool("POLYWEATHER_PREWARM_FORCE_REFRESH", False), + "base_url": str(os.getenv("POLYWEATHER_BACKEND_URL") or "http://127.0.0.1:8000").strip(), + } + validation_error = None + if not str(os.getenv("POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN") or "").strip(): + validation_error = "missing_POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN" + return self._start_with_validation( + key="dashboard_prewarm", + label="站点面板预热", + configured_enabled=enabled, + details=details, + validation_error=validation_error, + starter=lambda: import_module("src.utils.prewarm_dashboard").start_prewarm_worker_thread(), + ) + def _start_polygon_wallet_loop(self) -> LoopStatus: enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False) chat_ids = get_telegram_chat_ids_from_env() diff --git a/src/data_collection/metar_sources.py b/src/data_collection/metar_sources.py index fe24be34..174d22de 100644 --- a/src/data_collection/metar_sources.py +++ b/src/data_collection/metar_sources.py @@ -5,7 +5,7 @@ import time from datetime import datetime, timedelta, timezone from typing import Dict, List, Optional -import requests +import httpx from loguru import logger from src.utils.metrics import record_source_call @@ -209,7 +209,7 @@ class MetarSourceMixin: record_source_call("metar", "current", "success", (time.perf_counter() - started) * 1000.0) return result - except requests.exceptions.RequestException as exc: + except httpx.HTTPError as exc: logger.error(f"METAR 请求失败 ({icao}): {exc}") with self._metar_cache_lock: stale = self._metar_cache.get(cache_key) @@ -269,7 +269,7 @@ class MetarSourceMixin: self._taf_cache[cache_key] = {"d": result, "t": now_ts} record_source_call("taf", "current", "success", (time.perf_counter() - started) * 1000.0) return result - except requests.exceptions.RequestException as exc: + except httpx.HTTPError as exc: logger.error(f"TAF 请求失败 ({icao}): {exc}") with self._taf_cache_lock: stale = self._taf_cache.get(cache_key) diff --git a/src/data_collection/mgm_sources.py b/src/data_collection/mgm_sources.py index c035bcf5..b5000e35 100644 --- a/src/data_collection/mgm_sources.py +++ b/src/data_collection/mgm_sources.py @@ -32,7 +32,7 @@ class MgmSourceMixin: # 1. 实时数据 (添加时间戳防止 CDN 缓存) import time - obs_resp = self.session.get( + obs_resp = self._http_get( f"{base_url}/sondurumlar?istno={istno}&_={int(time.time() * 1000)}", headers=headers, timeout=self.timeout, @@ -95,7 +95,7 @@ class MgmSourceMixin: ] for forecast_url in forecast_urls: try: - daily_resp = self.session.get( + daily_resp = self._http_get( forecast_url, headers=headers, timeout=self.timeout ) if daily_resp.status_code == 200: @@ -132,7 +132,7 @@ class MgmSourceMixin: # 3. 小时预报 try: - hourly_resp = self.session.get( + hourly_resp = self._http_get( f"{base_url}/tahminler/saatlik?istno={istno}", headers=headers, timeout=self.timeout @@ -246,7 +246,11 @@ class MgmSourceMixin: try: # 1. 加载测站元数据 (缓存到实例中)，用于过滤属于该省份的站点 if not getattr(self, "mgm_stations_meta", None): - meta_resp = self.session.get(f"{base_url}/istasyonlar", headers=headers, timeout=self.timeout) + meta_resp = self._http_get( + f"{base_url}/istasyonlar", + headers=headers, + timeout=self.timeout, + ) if meta_resp.status_code == 200: meta_json = meta_resp.json() if isinstance(meta_json, list): @@ -293,7 +297,7 @@ class MgmSourceMixin: try: # sondurumlar?istno={ist_no} 是目前最稳的获取多站数据的办法 url = f"{base_url}/sondurumlar?istno={ist_no}&_={int(time.time() * 1000)}" - resp = self.session.get(url, headers=headers, timeout=5) + resp = self._http_get(url, headers=headers, timeout=5) if resp.status_code == 200: obs_list = resp.json() if obs_list: diff --git a/src/data_collection/nmc_sources.py b/src/data_collection/nmc_sources.py index 066958ed..22390a23 100644 --- a/src/data_collection/nmc_sources.py +++ b/src/data_collection/nmc_sources.py @@ -45,6 +45,20 @@ NMC_CITY_REFERENCES: Dict[str, Dict[str, Any]] = { class NmcSourceMixin: + def _nmc_http_get(self, url: str): + getter = getattr(self, "_http_get", None) + if callable(getter): + return getter(url) + return self.session.get(url, timeout=self.timeout) + + def _nmc_http_get_json(self, url: str): + getter = getattr(self, "_http_get_json", None) + if callable(getter): + return getter(url) + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.json() + @staticmethod def _nmc_optional_text(value: Any) -> Optional[str]: text = str(value or "").strip() @@ -73,7 +87,7 @@ class NmcSourceMixin: return None try: - resp = self.session.get(page_url, timeout=self.timeout) + resp = self._nmc_http_get(page_url) resp.raise_for_status() match = re.search( r"renderWeatherRealPanel\('([^']+)',\s*'([^']+)'\)", @@ -116,9 +130,7 @@ class NmcSourceMixin: try: url = f"https://www.nmc.cn/rest/real/{station_code}" - resp = self.session.get(url, timeout=self.timeout) - resp.raise_for_status() - payload = resp.json() + payload = self._nmc_http_get_json(url) if not isinstance(payload, dict) or not isinstance(payload.get("weather"), dict): record_source_call("nmc", "current", "empty", (time.perf_counter() - started) * 1000.0) return None diff --git a/src/data_collection/nws_open_meteo_sources.py b/src/data_collection/nws_open_meteo_sources.py index 0657e549..b225b40a 100644 --- a/src/data_collection/nws_open_meteo_sources.py +++ b/src/data_collection/nws_open_meteo_sources.py @@ -21,7 +21,7 @@ class NwsOpenMeteoSourceMixin: points_url = f"https://api.weather.gov/points/{lat},{lon}" headers = {"User-Agent": "PolyWeather/1.0 (weather-bot)"} - points_resp = self.session.get( + points_resp = self._http_get( points_url, headers=headers, timeout=self.timeout ) points_resp.raise_for_status() @@ -35,7 +35,7 @@ class NwsOpenMeteoSourceMixin: return None # 2. 获取预报 - forecast_resp = self.session.get( + forecast_resp = self._http_get( forecast_url, headers=headers, timeout=self.timeout ) forecast_resp.raise_for_status() @@ -48,7 +48,7 @@ class NwsOpenMeteoSourceMixin: hourly_periods = [] if hourly_url: - hourly_resp = self.session.get( + hourly_resp = self._http_get( hourly_url, headers=headers, timeout=self.timeout ) hourly_resp.raise_for_status() @@ -57,7 +57,7 @@ class NwsOpenMeteoSourceMixin: active_alerts = [] try: - alerts_resp = self.session.get( + alerts_resp = self._http_get( "https://api.weather.gov/alerts/active", params={"point": f"{lat},{lon}"}, headers=headers, @@ -205,7 +205,7 @@ class NwsOpenMeteoSourceMixin: params["temperature_unit"] = "celsius" self._wait_open_meteo_slot("forecast") - response = self.session.get( + response = self._http_get( url, params=params, timeout=self.timeout, @@ -366,7 +366,7 @@ class NwsOpenMeteoSourceMixin: params["temperature_unit"] = "celsius" self._wait_open_meteo_slot("ensemble") - response = self.session.get( + response = self._http_get( url, params=params, timeout=self.timeout, @@ -523,7 +523,7 @@ class NwsOpenMeteoSourceMixin: params["temperature_unit"] = "fahrenheit" self._wait_open_meteo_slot("multi-model") - response = self.session.get( + response = self._http_get( url, params=params, timeout=self.timeout, diff --git a/src/data_collection/polymarket_readonly.py b/src/data_collection/polymarket_readonly.py index d46816e6..efcbf26b 100644 --- a/src/data_collection/polymarket_readonly.py +++ b/src/data_collection/polymarket_readonly.py @@ -19,7 +19,7 @@ import unicodedata from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple -import requests +import httpx from loguru import logger from src.data_collection.city_registry import ALIASES, CITY_REGISTRY @@ -383,7 +383,10 @@ class PolymarketReadOnlyLayer: ) self.edge_threshold = _safe_float(os.getenv("POLYMARKET_SIGNAL_EDGE_PCT")) or 2.0 - self._session = requests.Session() + self._session = httpx.Client( + timeout=self.http_timeout, + follow_redirects=True, + ) self._markets_cache: Dict[str, Dict[str, Any]] = {} self._active_markets_cache: Dict[str, Any] = {"data": [], "t": 0.0} self._broad_markets_cache: Dict[str, Any] = {"data": [], "t": 0.0} diff --git a/src/data_collection/settlement_sources.py b/src/data_collection/settlement_sources.py index 78b7afef..4368be19 100644 --- a/src/data_collection/settlement_sources.py +++ b/src/data_collection/settlement_sources.py @@ -226,13 +226,13 @@ class SettlementSourceMixin: try: base = "https://data.weather.gov.hk/weatherAPI/hko_data/regional-weather" - temp_csv = self.session.get(f"{base}/latest_1min_temperature.csv", timeout=self.timeout) + temp_csv = self._http_get(f"{base}/latest_1min_temperature.csv", timeout=self.timeout) temp_csv.raise_for_status() - maxmin_csv = self.session.get(f"{base}/latest_since_midnight_maxmin.csv", timeout=self.timeout) + maxmin_csv = self._http_get(f"{base}/latest_since_midnight_maxmin.csv", timeout=self.timeout) maxmin_csv.raise_for_status() - humidity_csv = self.session.get(f"{base}/latest_1min_humidity.csv", timeout=self.timeout) + humidity_csv = self._http_get(f"{base}/latest_1min_humidity.csv", timeout=self.timeout) humidity_csv.raise_for_status() - wind_csv = self.session.get(f"{base}/latest_10min_wind.csv", timeout=self.timeout) + wind_csv = self._http_get(f"{base}/latest_10min_wind.csv", timeout=self.timeout) wind_csv.raise_for_status() temp_rows = self._csv_rows(temp_csv.text) @@ -302,7 +302,7 @@ class SettlementSourceMixin: try: url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/O-A0003-001" - response = self.session.get( + response = self._http_get( url, params={"Authorization": self.cwa_open_data_auth, "format": "JSON", "StationId": "466920"}, timeout=self.timeout, @@ -361,7 +361,7 @@ class SettlementSourceMixin: def fetch_hko_forecast(self) -> Optional[float]: try: url = "https://data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType=fnd&lang=tc" - res = self.session.get(url, timeout=self.timeout).json() + res = self._http_get_json(url, timeout=self.timeout) return float(res["weatherForecast"][0]["forecastMaxtemp"]["value"]) except Exception as exc: logger.warning(f"HKO Forecast request failed: {exc}") @@ -372,11 +372,11 @@ class SettlementSourceMixin: if not self.cwa_open_data_auth: return None url = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-D0047-061" - res = self.session.get( + res = self._http_get_json( url, params={"Authorization": self.cwa_open_data_auth, "format": "JSON", "elementName": "MaxT"}, timeout=self.timeout, - ).json() + ) locs = res.get("records", {}).get("Locations", [])[0].get("Location", []) if not locs: return None @@ -402,7 +402,7 @@ class SettlementSourceMixin: return cached try: - response = self.session.get( + response = self._http_get( "https://api.synopticdata.com/v2/stations/timeseries", params={ "STID": normalized_station_code, @@ -519,7 +519,7 @@ class SettlementSourceMixin: query = {"token": self.IMGW_METEO_API_TOKEN} if isinstance(params, dict): query.update(params) - response = self.session.get( + response = self._http_get( f"{self.IMGW_METEO_API_BASE}/{path}", params=query, timeout=self.timeout, diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 201ff413..cf96eb33 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -1,7 +1,8 @@ import os -import requests +import httpx import re import threading +import time from typing import Optional, Dict, List from datetime import datetime, timedelta from loguru import logger @@ -110,7 +111,17 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self.config = config self.timeout = 30 # 增加超时以支持高延迟 VPS - self.session = requests.Session() + self.http_retry_count = max( + 0, int(os.getenv("POLYWEATHER_HTTP_RETRY_COUNT", "1")) + ) + self.http_retry_backoff_sec = max( + 0.0, float(os.getenv("POLYWEATHER_HTTP_RETRY_BACKOFF_SEC", "0.35")) + ) + self.session = httpx.Client( + timeout=self.timeout, + follow_redirects=True, + limits=httpx.Limits(max_connections=50, max_keepalive_connections=20), + ) self.open_meteo_cache_ttl_sec = int( os.getenv("OPEN_METEO_CACHE_TTL_SEC", "900") ) @@ -186,11 +197,73 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if proxy: if not proxy.startswith("http"): proxy = f"http://{proxy}" - self.session.proxies = {"http": proxy, "https": proxy} + self.session = httpx.Client( + timeout=self.timeout, + follow_redirects=True, + proxy=proxy, + limits=httpx.Limits(max_connections=50, max_keepalive_connections=20), + ) logger.info(f"正在使用天气数据代理: {proxy}") logger.info("天气数据采集器初始化完成。") + @staticmethod + def _is_retryable_status(status_code: int) -> bool: + return status_code in {408, 500, 502, 503, 504} + + def _http_get(self, url: str, **kwargs) -> httpx.Response: + if "timeout" not in kwargs: + kwargs["timeout"] = self.timeout + last_exc: Optional[Exception] = None + last_response: Optional[httpx.Response] = None + attempts = self.http_retry_count + 1 + for attempt in range(attempts): + try: + response = self.session.get(url, **kwargs) + last_response = response + if ( + attempt < attempts - 1 + and self._is_retryable_status(response.status_code) + ): + wait_for = self.http_retry_backoff_sec * (attempt + 1) + logger.debug( + "HTTP GET retrying url={} status={} attempt={}/{} wait={}s", + url, + response.status_code, + attempt + 1, + attempts, + round(wait_for, 2), + ) + if wait_for > 0: + time.sleep(wait_for) + continue + return response + except (httpx.TimeoutException, httpx.NetworkError) as exc: + last_exc = exc + if attempt >= attempts - 1: + break + wait_for = self.http_retry_backoff_sec * (attempt + 1) + logger.debug( + "HTTP GET retrying url={} error={} attempt={}/{} wait={}s", + url, + type(exc).__name__, + attempt + 1, + attempts, + round(wait_for, 2), + ) + if wait_for > 0: + time.sleep(wait_for) + if last_exc is not None: + raise last_exc + if last_response is not None: + return last_response + raise RuntimeError(f"HTTP GET failed without response: {url}") + + def _http_get_json(self, url: str, **kwargs): + response = self._http_get(url, **kwargs) + response.raise_for_status() + return response.json() + def fetch_from_openweather(self, city: str, country: str = None) -> Optional[Dict]: """ Fetch current weather and forecast from OpenWeatherMap @@ -210,7 +283,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour try: # Current weather current_url = "https://api.openweathermap.org/data/2.5/weather" - current_response = self.session.get( + current_response = self._http_get( current_url, params={"q": query, "appid": self.openweather_key, "units": "metric"}, timeout=self.timeout, @@ -220,7 +293,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour # 5-day forecast forecast_url = "https://api.openweathermap.org/data/2.5/forecast" - forecast_response = self.session.get( + forecast_response = self._http_get( forecast_url, params={"q": query, "appid": self.openweather_key, "units": "metric"}, timeout=self.timeout, @@ -245,7 +318,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour "forecast": self._parse_openweather_forecast(forecast_data), } - except requests.exceptions.RequestException as e: + except httpx.HTTPError as e: logger.error(f"OpenWeatherMap request failed: {e}") return None @@ -290,7 +363,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour try: url = f"https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/{city}/{start_date}/{end_date}" - response = self.session.get( + response = self._http_get( url, params={ "unitGroup": "metric", @@ -322,7 +395,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ], } - except requests.exceptions.RequestException as e: + except httpx.HTTPError as e: logger.error(f"Visual Crossing request failed: {e}") return None @@ -399,7 +472,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour try: url = "https://geocoding-api.open-meteo.com/v1/search" - response = self.session.get( + response = self._http_get( url, params={"name": city, "count": 1, "language": "en", "format": "json"}, timeout=15, # 增加超时时间到 15s diff --git a/src/utils/prewarm_dashboard.py b/src/utils/prewarm_dashboard.py new file mode 100644 index 00000000..23aa828a --- /dev/null +++ b/src/utils/prewarm_dashboard.py @@ -0,0 +1,305 @@ +from __future__ import annotations + +import json +import os +import random +import threading +import time +from datetime import datetime +from typing import Any, Dict, Optional + +import httpx +from loguru import logger + + +DEFAULT_CITIES = [ + "ankara", + "istanbul", + "shanghai", + "beijing", + "shenzhen", + "wuhan", + "chengdu", + "chongqing", + "hong kong", + "taipei", + "london", + "paris", + "new york", + "los angeles", +] + +_RUNTIME_LOCK = threading.Lock() +_WORKER_THREAD: Optional[threading.Thread] = None +_RUNTIME_STATE: Dict[str, Any] = { + "cycle_count": 0, + "success_count": 0, + "failure_count": 0, + "last_started_at": None, + "last_finished_at": None, + "last_duration_sec": None, + "last_success": None, + "last_http_status": None, + "last_error": None, + "last_requested_cities": [], + "last_requested_city_count": 0, + "last_include_detail": False, + "last_include_market": False, + "last_force_refresh": False, + "last_warmed_count": 0, + "last_summary_ok": 0, + "last_detail_ok": 0, + "last_market_ok": 0, + "last_failed_count": 0, +} + + +def _truthy_env(value: str, default: bool = False) -> bool: + raw = str(os.getenv(value) or "").strip().lower() + if not raw: + return default + return raw in {"1", "true", "yes", "on"} + + +def _parse_cities(value: str) -> list[str]: + items = [item.strip() for item in str(value or "").split(",")] + return [item for item in items if item] + + +def _update_runtime_state(**kwargs: Any) -> None: + with _RUNTIME_LOCK: + _RUNTIME_STATE.update(kwargs) + + +def _record_prewarm_result( + *, + ok: bool, + duration_sec: float, + http_status: Optional[int], + error: Optional[str], + warmed_count: int, + summary_ok: int, + detail_ok: int, + market_ok: int, + failed_count: int, +) -> None: + with _RUNTIME_LOCK: + _RUNTIME_STATE["cycle_count"] = int(_RUNTIME_STATE.get("cycle_count") or 0) + 1 + if ok: + _RUNTIME_STATE["success_count"] = int(_RUNTIME_STATE.get("success_count") or 0) + 1 + else: + _RUNTIME_STATE["failure_count"] = int(_RUNTIME_STATE.get("failure_count") or 0) + 1 + _RUNTIME_STATE["last_finished_at"] = datetime.now().isoformat(timespec="seconds") + _RUNTIME_STATE["last_duration_sec"] = round(float(duration_sec), 2) + _RUNTIME_STATE["last_success"] = bool(ok) + _RUNTIME_STATE["last_http_status"] = http_status + _RUNTIME_STATE["last_error"] = error + _RUNTIME_STATE["last_warmed_count"] = int(warmed_count or 0) + _RUNTIME_STATE["last_summary_ok"] = int(summary_ok or 0) + _RUNTIME_STATE["last_detail_ok"] = int(detail_ok or 0) + _RUNTIME_STATE["last_market_ok"] = int(market_ok or 0) + _RUNTIME_STATE["last_failed_count"] = int(failed_count or 0) + + +def get_prewarm_runtime_summary() -> Dict[str, Any]: + configured_cities = _parse_cities(str(os.getenv("POLYWEATHER_PREWARM_CITIES") or ",".join(DEFAULT_CITIES))) + with _RUNTIME_LOCK: + runtime = dict(_RUNTIME_STATE) + return { + "enabled": _truthy_env("POLYWEATHER_DASHBOARD_PREWARM_ENABLED", False), + "base_url": str(os.getenv("POLYWEATHER_BACKEND_URL") or "http://127.0.0.1:8000").strip(), + "configured_cities": configured_cities, + "configured_city_count": len(configured_cities), + "interval_sec": max(30, int(os.getenv("POLYWEATHER_PREWARM_INTERVAL_SEC", "300"))), + "jitter_sec": max(0, int(os.getenv("POLYWEATHER_PREWARM_JITTER_SEC", "20"))), + "include_detail": _truthy_env("POLYWEATHER_PREWARM_INCLUDE_DETAIL", True), + "include_market": _truthy_env("POLYWEATHER_PREWARM_INCLUDE_MARKET", True), + "force_refresh": _truthy_env("POLYWEATHER_PREWARM_FORCE_REFRESH", False), + "thread_alive": bool(_WORKER_THREAD and _WORKER_THREAD.is_alive()), + "runtime": runtime, + } + + +def run_prewarm( + *, + base_url: str, + cities: str, + force_refresh: bool, + include_detail: bool, + include_market: bool, +) -> int: + token = str(os.getenv("POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN") or "").strip() + requested_cities = _parse_cities(cities) + started = time.perf_counter() + _update_runtime_state( + last_started_at=datetime.now().isoformat(timespec="seconds"), + last_requested_cities=requested_cities, + last_requested_city_count=len(requested_cities), + last_include_detail=bool(include_detail), + last_include_market=bool(include_market), + last_force_refresh=bool(force_refresh), + last_error=None, + last_http_status=None, + ) + if not token: + _record_prewarm_result( + ok=False, + duration_sec=time.perf_counter() - started, + http_status=None, + error="missing_backend_token", + warmed_count=0, + summary_ok=0, + detail_ok=0, + market_ok=0, + failed_count=0, + ) + print( + json.dumps( + { + "ok": False, + "reason": "missing_backend_token", + "detail": "POLYWEATHER_BACKEND_ENTITLEMENT_TOKEN is required", + }, + ensure_ascii=False, + ) + ) + return 1 + + try: + with httpx.Client(timeout=180, follow_redirects=True) as client: + response = client.post( + f"{base_url.rstrip('/')}/api/system/prewarm", + params={ + "cities": cities, + "force_refresh": str(bool(force_refresh)).lower(), + "include_detail": str(bool(include_detail)).lower(), + "include_market": str(bool(include_market)).lower(), + }, + headers={ + "Accept": "application/json", + "x-polyweather-entitlement": token, + }, + ) + try: + payload = response.json() + except Exception: + payload = {"ok": False, "raw": (response.text or "")[:500]} + warmed_count = int((payload or {}).get("count") or 0) if isinstance(payload, dict) else 0 + summary_ok = int((payload or {}).get("summary_ok") or 0) if isinstance(payload, dict) else 0 + detail_ok = int((payload or {}).get("detail_ok") or 0) if isinstance(payload, dict) else 0 + market_ok = int((payload or {}).get("market_ok") or 0) if isinstance(payload, dict) else 0 + failed_count = int((payload or {}).get("failed_count") or 0) if isinstance(payload, dict) else 0 + ok = bool(response.is_success and (not isinstance(payload, dict) or payload.get("ok", True))) + _record_prewarm_result( + ok=ok, + duration_sec=time.perf_counter() - started, + http_status=response.status_code, + error=None if ok else str((payload or {}).get("detail") or (payload or {}).get("reason") or response.text[:200]), + warmed_count=warmed_count, + summary_ok=summary_ok, + detail_ok=detail_ok, + market_ok=market_ok, + failed_count=failed_count, + ) + print(json.dumps(payload, ensure_ascii=False, indent=2)) + return 0 if response.is_success else 1 + except Exception as exc: + _record_prewarm_result( + ok=False, + duration_sec=time.perf_counter() - started, + http_status=None, + error=str(exc), + warmed_count=0, + summary_ok=0, + detail_ok=0, + market_ok=0, + failed_count=0, + ) + print(json.dumps({"ok": False, "reason": "request_failed", "detail": str(exc)}, ensure_ascii=False, indent=2)) + return 1 + + +def run_worker_loop( + *, + base_url: str, + cities: str, + interval_sec: int, + jitter_sec: int, + force_refresh: bool, + include_detail: bool, + include_market: bool, + once: bool = False, +) -> int: + interval_sec = max(30, int(interval_sec)) + jitter_sec = max(0, int(jitter_sec)) + + logger.info( + "dashboard prewarm worker started base_url={} cities={} interval_sec={} jitter_sec={} include_detail={} include_market={} force_refresh={} once={}", + base_url, + cities, + interval_sec, + jitter_sec, + bool(include_detail), + bool(include_market), + bool(force_refresh), + bool(once), + ) + + while True: + started = time.perf_counter() + exit_code = run_prewarm( + base_url=base_url, + cities=cities, + force_refresh=bool(force_refresh), + include_detail=bool(include_detail), + include_market=bool(include_market), + ) + elapsed = time.perf_counter() - started + logger.info( + "dashboard prewarm worker cycle exit_code={} elapsed_sec={} finished_at={}", + exit_code, + round(elapsed, 2), + datetime.now().isoformat(timespec="seconds"), + ) + if once: + return exit_code + + sleep_sec = max(5.0, interval_sec - elapsed) + if jitter_sec > 0: + sleep_sec += random.randint(0, jitter_sec) + logger.info("dashboard prewarm worker sleeping sleep_sec={}", round(sleep_sec, 2)) + time.sleep(sleep_sec) + + +def start_prewarm_worker_thread() -> Optional[threading.Thread]: + enabled = str(os.getenv("POLYWEATHER_DASHBOARD_PREWARM_ENABLED") or "").strip().lower() + if enabled not in {"1", "true", "yes", "on"}: + return None + + base_url = str(os.getenv("POLYWEATHER_BACKEND_URL") or "http://127.0.0.1:8000").strip() + cities = str(os.getenv("POLYWEATHER_PREWARM_CITIES") or ",".join(DEFAULT_CITIES)).strip() + interval_sec = int(os.getenv("POLYWEATHER_PREWARM_INTERVAL_SEC", "300")) + jitter_sec = int(os.getenv("POLYWEATHER_PREWARM_JITTER_SEC", "20")) + force_refresh = str(os.getenv("POLYWEATHER_PREWARM_FORCE_REFRESH") or "").strip().lower() in {"1", "true", "yes", "on"} + include_detail = str(os.getenv("POLYWEATHER_PREWARM_INCLUDE_DETAIL", "true")).strip().lower() in {"1", "true", "yes", "on"} + include_market = str(os.getenv("POLYWEATHER_PREWARM_INCLUDE_MARKET", "true")).strip().lower() in {"1", "true", "yes", "on"} + + thread = threading.Thread( + target=run_worker_loop, + kwargs={ + "base_url": base_url, + "cities": cities, + "interval_sec": interval_sec, + "jitter_sec": jitter_sec, + "force_refresh": force_refresh, + "include_detail": include_detail, + "include_market": include_market, + "once": False, + }, + name="dashboard-prewarm-worker", + daemon=True, + ) + thread.start() + global _WORKER_THREAD + _WORKER_THREAD = thread + return thread diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 24126252..996af7d9 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -28,16 +28,22 @@ def test_system_status_returns_summary_shape(): assert 'features' in payload assert 'integrations' in payload assert 'cache' in payload + assert 'analysis' in payload['cache'] assert 'probability' in payload assert 'rollout' in payload['probability'] assert payload['probability']['rollout']['decision']['decision'] in {'hold', 'observe', 'promote'} assert 'training_data' in payload assert 'station_networks' in payload + assert 'prewarm' in payload assert 'truth_records' in payload['training_data'] assert 'training_features' in payload['training_data'] assert 'city_coverage' in payload['training_data'] assert 'model_city_coverage' in payload['training_data'] assert 'artifacts' in payload['training_data'] + assert 'metar_entries' in payload['cache'] + assert 'nmc_entries' in payload['cache'] + assert 'runtime' in payload['prewarm'] + assert 'last_summary_ok' in payload['prewarm']['runtime'] assert 'cities_count' in payload diff --git a/web/analysis_service.py b/web/analysis_service.py index e31590af..7b4f2305 100644 --- a/web/analysis_service.py +++ b/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"] diff --git a/web/core.py b/web/core.py index 51d6af14..bec62ea5 100644 --- a/web/core.py +++ b/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), } diff --git a/web/routes.py b/web/routes.py index 999a1db8..12d98188 100644 --- a/web/routes.py +++ b/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, )