From 15de44767e9fd6d25d2196aa12beecd68053aa46 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 30 Jun 2026 12:40:40 +0800 Subject: [PATCH] Batch scan terminal model forecasts --- frontend/app/api/scan/terminal/route.ts | 23 +- .../__tests__/proxyCachePolicy.test.ts | 32 +- .../__tests__/apiPerformanceTiming.test.ts | 6 +- scripts/configure_cloudflare_free.py | 5 +- scripts/validate_frontend_cache.sh | 1 - tests/test_api_performance_timing.py | 4 +- tests/test_cloudflare_cache_validation.py | 2 +- tests/test_cloudflare_free_config.py | 6 +- tests/test_deployment_runtime_config.py | 2 +- tests/test_scan_terminal_modules.py | 277 +++++++++++++++++ web/routers/scan.py | 15 +- web/scan_terminal_cache.py | 7 +- web/scan_terminal_city_row.py | 283 +++++++++++++++++- web/scan_terminal_service.py | 139 ++++++--- 14 files changed, 668 insertions(+), 134 deletions(-) diff --git a/frontend/app/api/scan/terminal/route.ts b/frontend/app/api/scan/terminal/route.ts index a4bcfc9a..3dfc742c 100644 --- a/frontend/app/api/scan/terminal/route.ts +++ b/frontend/app/api/scan/terminal/route.ts @@ -1,10 +1,8 @@ import { NextRequest, NextResponse } from "next/server"; import { proxyBackendJsonGet } from "@/lib/api-proxy"; import { - buildForceRefreshProxyCachePolicy, - buildScanTerminalResponseCacheControl, + NO_STORE_CACHE_CONTROL, } from "@/lib/proxy-cache-policy"; -import { DASHBOARD_REFRESH_POLICY_SEC } from "@/lib/refresh-policy"; import { createProxyTimer, finishProxyTimedResponse, @@ -12,10 +10,10 @@ import { const API_BASE = process.env.POLYWEATHER_API_BASE_URL; const SCAN_TERMINAL_PROXY_TIMEOUT_MS = Number( - process.env.POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "35000", + process.env.POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "60000", ); -export const maxDuration = 45; +export const maxDuration = 70; export async function GET(req: NextRequest) { const timer = createProxyTimer(req, "scan_terminal"); @@ -31,7 +29,6 @@ export async function GET(req: NextRequest) { } const params = new URLSearchParams(); - const forceRefresh = req.nextUrl.searchParams.get("force_refresh") ?? "false"; for (const key of [ "scan_mode", "min_price", @@ -56,11 +53,6 @@ export async function GET(req: NextRequest) { if (tradingRegion != null && tradingRegion !== "") { params.set("region", tradingRegion); } - const cachePolicy = buildForceRefreshProxyCachePolicy( - forceRefresh, - DASHBOARD_REFRESH_POLICY_SEC.scanRows, - ); - const url = `${API_BASE}/api/scan/terminal?${params.toString()}`; const controller = new AbortController(); @@ -68,15 +60,10 @@ export async function GET(req: NextRequest) { try { return await proxyBackendJsonGet(req, { - cacheControl: cachePolicy.responseCacheControl, - cacheControlForData: (data) => - buildScanTerminalResponseCacheControl( - data, - cachePolicy.responseCacheControl, - ), + cacheControl: NO_STORE_CACHE_CONTROL, + cacheControlForData: () => NO_STORE_CACHE_CONTROL, fetchCache: "no-store", publicMessage: "Failed to fetch scan terminal data", - revalidateSeconds: cachePolicy.revalidateSeconds, includeSupabaseIdentity: true, signal: controller.signal, timeoutPublicMessage: "Scan terminal request timed out", diff --git a/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts index 98ea1081..82f3386f 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts @@ -3,7 +3,6 @@ import fs from "node:fs"; import path from "node:path"; import { buildCityListCacheControl, - buildScanTerminalResponseCacheControl, buildCityDetailProxyCachePolicy, buildForceRefreshProxyCachePolicy, buildPublicEdgeCacheControl, @@ -40,42 +39,19 @@ export function runTests() { const scanForced = buildForceRefreshProxyCachePolicy("true", 10); assert.equal(scanForced.fetchMode, "no-store"); - const normalScanCache = "public, max-age=0, s-maxage=300, stale-while-revalidate=900"; - assert.equal( - buildScanTerminalResponseCacheControl({ status: "ready", stale: false }, normalScanCache), - normalScanCache, - ); - assert.equal( - buildScanTerminalResponseCacheControl({ status: "failed", stale: false }, normalScanCache), - NO_STORE_CACHE_CONTROL, - ); - assert.equal( - buildScanTerminalResponseCacheControl({ status: "partial", stale: false }, normalScanCache), - NO_STORE_CACHE_CONTROL, - ); - assert.equal( - buildScanTerminalResponseCacheControl({ status: "ready", stale: true }, normalScanCache), - NO_STORE_CACHE_CONTROL, - ); - const scanTerminalProxySource = fs.readFileSync( path.join(process.cwd(), "app", "api", "scan", "terminal", "route.ts"), "utf8", ); - assert.match( - scanTerminalProxySource, - /DASHBOARD_REFRESH_POLICY_SEC\.scanRows/, - "scan terminal proxy cache TTL should match the dashboard scan refresh cadence instead of a short literal TTL", - ); assert.doesNotMatch( scanTerminalProxySource, - /buildForceRefreshProxyCachePolicy\(forceRefresh,\s*10\)/, - "scan terminal proxy must not use the old 10 second edge cache because it over-drives the slow scan endpoint", + /buildForceRefreshProxyCachePolicy/, + "scan terminal proxy must not use public edge cache because rows depend on user entitlement and live scan state", ); assert.match( scanTerminalProxySource, - /cacheControlForData:\s*\(data\)\s*=>\s*buildScanTerminalResponseCacheControl/, - "scan terminal proxy must not CDN-cache failed, stale, or partial business payloads", + /cacheControlForData:\s*\(\)\s*=>\s*NO_STORE_CACHE_CONTROL/, + "scan terminal proxy must keep all scan payloads out of shared CDN caches", ); assert.match( scanTerminalProxySource, diff --git a/frontend/components/ops/__tests__/apiPerformanceTiming.test.ts b/frontend/components/ops/__tests__/apiPerformanceTiming.test.ts index 1b60fb8d..3b990613 100644 --- a/frontend/components/ops/__tests__/apiPerformanceTiming.test.ts +++ b/frontend/components/ops/__tests__/apiPerformanceTiming.test.ts @@ -117,12 +117,12 @@ export function runTests() { assert.match(scanTerminalProxy, /timing:\s*timer/); assert.match( scanTerminalProxy, - /POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS\s*\|\|\s*"35000"/, - "scan terminal proxy should allow the production backend enough time to return before the 45 second route cap", + /POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS\s*\|\|\s*"60000"/, + "scan terminal proxy should allow the production backend enough time for batched model refreshes", ); assert.match( scanTerminalProxy, - /export const maxDuration = 45/, + /export const maxDuration = 70/, "scan terminal proxy timeout budget should remain below the Next route execution cap", ); diff --git a/scripts/configure_cloudflare_free.py b/scripts/configure_cloudflare_free.py index 205c9e5f..8fdd12b8 100644 --- a/scripts/configure_cloudflare_free.py +++ b/scripts/configure_cloudflare_free.py @@ -64,8 +64,7 @@ SCAN_EXPRESSION = """( http.host eq "polyweather.top" and http.request.method in {"GET" "HEAD"} and ( - http.request.uri.path eq "/api/scan/terminal" - or http.request.uri.path eq "/api/system/status" + http.request.uri.path eq "/api/system/status" ) )""" @@ -135,7 +134,7 @@ def build_managed_cache_rules() -> List[Dict[str, Any]]: _cache_rule("pages", "PolyWeather: cache public pages for ten minutes", PUBLIC_PAGES_EXPRESSION, 600), _cache_rule("cities", "PolyWeather: cache the public city list for five minutes", CITIES_EXPRESSION, 300), _cache_rule("city_detail", "PolyWeather: cache public city detail when the origin allows it", CITY_DETAIL_EXPRESSION), - _cache_rule("scan", "PolyWeather: cache scan and system status when the origin allows it", SCAN_EXPRESSION), + _cache_rule("system_status", "PolyWeather: cache system status when the origin allows it", SCAN_EXPRESSION), { "ref": f"{MANAGED_RULE_REF_PREFIX}bypass", "description": "PolyWeather: bypass backend, sensitive, realtime, and force-refresh requests", diff --git a/scripts/validate_frontend_cache.sh b/scripts/validate_frontend_cache.sh index e84955a0..9307dcf3 100644 --- a/scripts/validate_frontend_cache.sh +++ b/scripts/validate_frontend_cache.sh @@ -253,7 +253,6 @@ main() { check_cached_endpoint "/api/history/ankara" "history" check_if_none_match_304 "/api/history/ankara" "history" - check_cloudflare_cache_hit "/api/scan/terminal?limit=1" "scan terminal edge cache" print_line "" if [ "$FAIL_COUNT" -gt 0 ]; then diff --git a/tests/test_api_performance_timing.py b/tests/test_api_performance_timing.py index 728b3323..c7d54ad6 100644 --- a/tests/test_api_performance_timing.py +++ b/tests/test_api_performance_timing.py @@ -144,9 +144,7 @@ def test_scan_terminal_response_includes_backend_server_timing(monkeypatch): assert "scan_terminal_assert_entitlement" in server_timing assert "scan_terminal_build_payload" in server_timing assert "scan_terminal_total" in server_timing - assert response.headers["cache-control"] == ( - "public, max-age=0, s-maxage=300, stale-while-revalidate=900" - ) + assert response.headers["cache-control"] == "no-store, max-age=0" assert response.headers["cloudflare-cdn-cache-control"] == response.headers["cache-control"] diff --git a/tests/test_cloudflare_cache_validation.py b/tests/test_cloudflare_cache_validation.py index 4e10acce..da784156 100644 --- a/tests/test_cloudflare_cache_validation.py +++ b/tests/test_cloudflare_cache_validation.py @@ -14,4 +14,4 @@ def test_frontend_cache_validator_checks_cloudflare_edge_hits(): assert "MISS" in script assert "REVALIDATED" in script assert 'check_cloudflare_cache_hit "/api/cities" "cities edge cache"' in script - assert 'check_cloudflare_cache_hit "/api/scan/terminal?limit=1" "scan terminal edge cache"' in script + assert "/api/scan/terminal?limit=1" not in script diff --git a/tests/test_cloudflare_free_config.py b/tests/test_cloudflare_free_config.py index 154de64e..1cb9ef31 100644 --- a/tests/test_cloudflare_free_config.py +++ b/tests/test_cloudflare_free_config.py @@ -14,7 +14,7 @@ def test_cloudflare_managed_rules_apply_path_specific_edge_ttls_then_bypass_sens f"{MANAGED_RULE_REF_PREFIX}pages", f"{MANAGED_RULE_REF_PREFIX}cities", f"{MANAGED_RULE_REF_PREFIX}city_detail", - f"{MANAGED_RULE_REF_PREFIX}scan", + f"{MANAGED_RULE_REF_PREFIX}system_status", f"{MANAGED_RULE_REF_PREFIX}bypass", ] assert [ @@ -31,6 +31,8 @@ def test_cloudflare_managed_rules_apply_path_specific_edge_ttls_then_bypass_sens "cache": True, "browser_ttl": {"mode": "respect_origin"}, } + assert 'http.request.uri.path eq "/api/system/status"' in rules[4]["expression"] + assert "/api/scan/terminal" not in rules[4]["expression"] assert rules[-1]["action_parameters"]["cache"] is False assert 'http.host eq "api.polyweather.top"' in rules[-1]["expression"] assert 'http.request.uri.query contains "force_refresh=true"' in rules[-1]["expression"] @@ -64,7 +66,7 @@ def test_cloudflare_rule_merge_preserves_unmanaged_rules_and_puts_bypass_last(): f"{MANAGED_RULE_REF_PREFIX}pages", f"{MANAGED_RULE_REF_PREFIX}cities", f"{MANAGED_RULE_REF_PREFIX}city_detail", - f"{MANAGED_RULE_REF_PREFIX}scan", + f"{MANAGED_RULE_REF_PREFIX}system_status", f"{MANAGED_RULE_REF_PREFIX}bypass", ] assert merged[-1]["action_parameters"]["cache"] is False diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 45060ead..62d8cdab 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -193,7 +193,7 @@ def test_scan_terminal_backend_timeout_returns_before_next_proxy_abort(): ROOT / "web" / "services" / "scan_terminal_config.py" ).read_text(encoding="utf-8") - assert 'POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "35000"' in route_source + assert 'POLYWEATHER_SCAN_TERMINAL_PROXY_TIMEOUT_MS || "60000"' in route_source assert '"POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC",\n 30,' in config_source assert '"POLYWEATHER_SCAN_TERMINAL_MAX_WORKERS",\n 1,' in config_source assert ( diff --git a/tests/test_scan_terminal_modules.py b/tests/test_scan_terminal_modules.py index 5c8a55e6..b5013422 100644 --- a/tests/test_scan_terminal_modules.py +++ b/tests/test_scan_terminal_modules.py @@ -59,6 +59,19 @@ def test_scan_terminal_cache_hydrates_success_payload_from_redis(monkeypatch): assert cached["summary"]["candidate_total"] == 1 +def test_scan_terminal_redis_cache_prefix_skips_legacy_blank_snapshots(monkeypatch): + monkeypatch.delenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX", raising=False) + + assert scan_terminal_cache._redis_cache_prefix() == "polyweather:scan_terminal:v2:" + + monkeypatch.setenv( + "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX", + "polyweather:scan_terminal:v1:", + ) + + assert scan_terminal_cache._redis_cache_prefix() == "polyweather:scan_terminal:v2:" + + def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch): fake_redis = _FakeRedis() monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true") @@ -433,6 +446,270 @@ def test_scan_city_terminal_rows_refreshes_models_for_wrong_local_date_panel(mon ] +def test_scan_terminal_fetches_multi_model_daily_in_batches(monkeypatch): + today_paris = _local_date_for_offset(3600) + today_houston = _local_date_for_offset(-18000) + monkeypatch.setattr( + scan_terminal_city_row, + "CITIES", + { + "paris": {"lat": 48.85, "lon": 2.35, "tz": 3600}, + "houston": {"lat": 29.76, "lon": -95.36, "tz": -18000, "f": True}, + }, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_multi_model_cache", + {}, + raising=False, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_maybe_reload_open_meteo_disk_cache", + lambda: None, + raising=False, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_flush_open_meteo_disk_cache", + lambda: None, + raising=False, + ) + + requests = [] + + class _Response: + def __init__(self, payload): + self._payload = payload + + @staticmethod + def raise_for_status(): + return None + + def json(self): + return self._payload + + def _http_get(_url, *, params, timeout): + requests.append((params, timeout)) + unit_is_f = params.get("temperature_unit") == "fahrenheit" + date = today_houston if unit_is_f else today_paris + value = 94.0 if unit_is_f else 24.0 + return _Response( + [ + { + "daily": { + "time": [date], + "temperature_2m_max_ecmwf_ifs025": [value], + "temperature_2m_max_gfs_seamless": [value + 1], + } + } + ] + ) + + monkeypatch.setattr(scan_terminal_city_row._weather, "_http_get", _http_get, raising=False) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_wait_open_meteo_slot", + lambda _endpoint: None, + raising=False, + ) + + result = scan_terminal_city_row._fetch_scan_terminal_multi_model_batch(["paris", "houston"]) + + assert len(requests) == 2 + assert requests[0][0]["latitude"] == "48.85" + assert requests[1][0]["temperature_unit"] == "fahrenheit" + assert result["paris"]["daily_forecasts"][today_paris]["ECMWF"] == 24.0 + assert result["paris"]["daily_forecasts"][today_paris]["GFS"] == 25.0 + assert result["houston"]["daily_forecasts"][today_houston]["ECMWF"] == 94.0 + + +def test_scan_terminal_multi_model_batch_chunks_long_city_lists(monkeypatch): + today = _local_date_for_offset(0) + cities = { + f"city{i}": {"lat": float(i), "lon": float(i + 1), "tz": 0} + for i in range(45) + } + monkeypatch.setattr(scan_terminal_city_row, "CITIES", cities) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_multi_model_cache", + {}, + raising=False, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_maybe_reload_open_meteo_disk_cache", + lambda: None, + raising=False, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_flush_open_meteo_disk_cache", + lambda: None, + raising=False, + ) + monkeypatch.setattr( + scan_terminal_city_row._weather, + "_wait_open_meteo_slot", + lambda _endpoint: None, + raising=False, + ) + requests = [] + + class _Response: + def __init__(self, count): + self.count = count + + @staticmethod + def raise_for_status(): + return None + + def json(self): + return [ + { + "daily": { + "time": [today], + "temperature_2m_max_ecmwf_ifs025": [20.0 + idx], + } + } + for idx in range(self.count) + ] + + def _http_get(_url, *, params, timeout): + latitude_count = len(str(params["latitude"]).split(",")) + requests.append(latitude_count) + return _Response(latitude_count) + + monkeypatch.setattr(scan_terminal_city_row._weather, "_http_get", _http_get, raising=False) + + result = scan_terminal_city_row._fetch_scan_terminal_multi_model_batch(list(cities.keys())) + + assert requests == [20, 20, 5] + assert len(result) == 45 + assert result["city0"]["daily_forecasts"][today]["ECMWF"] == 20.0 + assert result["city44"]["daily_forecasts"][today]["ECMWF"] == 24.0 + + +def test_scan_terminal_uncached_passes_batch_multi_model_overrides(monkeypatch): + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + monkeypatch.setattr( + scan_terminal_service, + "CITIES", + {"paris": {"tz": 3600}, "houston": {"tz": -18000}}, + ) + monkeypatch.setattr(scan_terminal_service, "SCAN_TERMINAL_MAX_WORKERS", 2) + monkeypatch.setattr( + scan_terminal_service, + "_fetch_scan_terminal_multi_model_batch", + lambda cities: {"paris": {"source": "batch-paris"}}, + ) + seen = [] + + def _scan_city( + city, + _filters, + *, + force_refresh=False, + multi_model_override=None, + allow_direct_fetch=True, + ): + seen.append((city, force_refresh, multi_model_override, allow_direct_fetch)) + return { + "city": city, + "candidate_total": 1, + "primary_scores": [1.0], + "rows": [ + { + "id": f"{city}:today", + "market_key": f"{city}:today", + "final_score": 1.0, + "edge_percent": 0.0, + "volume": 0.0, + } + ], + } + + monkeypatch.setattr(scan_terminal_service, "_scan_city_terminal_rows", _scan_city) + + payload = scan_terminal_service._build_scan_terminal_payload_uncached( + {"limit": 2}, + force_refresh=True, + timeout_sec=5, + ) + + assert payload["status"] == "ready" + assert sorted(seen) == [ + ("paris", True, {"source": "batch-paris"}, False), + ] + assert payload["summary"]["total_city_count"] == 2 + assert payload["summary"]["scanned_city_count"] == 1 + + +def test_scan_city_terminal_rows_builds_forecast_only_row_from_batch_override(monkeypatch): + today = _local_date_for_offset(3600) + + class _Cache: + @staticmethod + def get_city_cache(*_args, **_kwargs): + raise AssertionError("batch forecast rows should not read panel cache") + + @staticmethod + def get_canonical_temperature(*_args, **_kwargs): + raise AssertionError("batch forecast rows should not read canonical cache") + + monkeypatch.setattr(scan_terminal_city_row, "_PANEL_CACHE_DB", _Cache()) + monkeypatch.setattr( + scan_terminal_city_row, + "calculate_deb_prediction", + lambda city, forecasts, raw_calculator=None: {"prediction": 24.5}, + ) + + result = scan_terminal_city_row._scan_city_terminal_rows( + "paris", + {"market_type": "maxtemp"}, + force_refresh=True, + multi_model_override={ + "daily_forecasts": {today: {"ECMWF": 24.0, "GFS": 25.0}}, + "dates": [today], + "unit": "celsius", + }, + allow_direct_fetch=False, + ) + + row = result["rows"][0] + assert row["local_date"] == today + assert row["deb_prediction"] == 24.5 + assert row["model_cluster_sources"] == {"ECMWF": 24.0, "GFS": 25.0} + assert row["forecast_refreshed"] is True + + +def test_scan_timeout_prefers_partial_rows_with_models_over_blank_stale_cache(monkeypatch): + cached_entry = { + "success_payload": { + "rows": [ + {"id": "old-1", "model_cluster_sources": {}}, + {"id": "old-2", "model_cluster_sources": {}}, + ] + }, + "last_failed_at": None, + } + + stale = scan_terminal_service._build_stale_payload_for_timeout_if_better_cached( + filters={"limit": 2}, + cached_entry=cached_entry, + ranked_rows=[ + { + "id": "new-1", + "model_cluster_sources": {"ECMWF": 24.0}, + } + ], + timeout_message="scan terminal build timed out after 30s", + ) + + assert stale is None + + def test_scan_city_terminal_rows_uses_canonical_without_analyze(monkeypatch): enqueued = [] diff --git a/web/routers/scan.py b/web/routers/scan.py index 3244d4a8..b67df189 100644 --- a/web/routers/scan.py +++ b/web/routers/scan.py @@ -5,7 +5,7 @@ from __future__ import annotations from fastapi import APIRouter, Request from fastapi.responses import JSONResponse -from web.services.cache_headers import NO_STORE_CACHE_CONTROL, public_edge_cache_control +from web.services.cache_headers import NO_STORE_CACHE_CONTROL from web.services.scan_api import ( get_scan_terminal_overview_payload, get_scan_terminal_payload, @@ -14,9 +14,6 @@ from web.services.request_timing import attach_server_timing_header router = APIRouter(tags=["scan"]) -SCAN_TERMINAL_CACHE_CONTROL = public_edge_cache_control(300, 900) - - @router.get("/api/scan/terminal") async def scan_terminal( request: Request, @@ -53,17 +50,11 @@ async def scan_terminal( region=region or trading_region or None, timezone_offset_seconds=timezone_offset_seconds, ) - status = str(payload.get("status") or "").strip().lower() - cache_control = ( - SCAN_TERMINAL_CACHE_CONTROL - if not force_refresh and status == "ready" and payload.get("stale") is not True - else NO_STORE_CACHE_CONTROL - ) response = JSONResponse( content=payload, headers={ - "Cache-Control": cache_control, - "Cloudflare-CDN-Cache-Control": cache_control, + "Cache-Control": NO_STORE_CACHE_CONTROL, + "Cloudflare-CDN-Cache-Control": NO_STORE_CACHE_CONTROL, }, ) attach_server_timing_header(response, request, "scan_terminal_server_timing") diff --git a/web/scan_terminal_cache.py b/web/scan_terminal_cache.py index 8da451e4..d8b79bca 100644 --- a/web/scan_terminal_cache.py +++ b/web/scan_terminal_cache.py @@ -43,10 +43,13 @@ def _redis_cache_ttl_sec() -> int: def _redis_cache_prefix() -> str: - return os.getenv( + prefix = os.getenv( "POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_PREFIX", - "polyweather:scan_terminal:v1:", + "polyweather:scan_terminal:v2:", ) + if prefix.endswith(":v1:"): + return f"{prefix[:-4]}:v2:" + return prefix def _redis_entry_key(cache_key: str) -> str: diff --git a/web/scan_terminal_city_row.py b/web/scan_terminal_city_row.py index 441e1559..18c4fc5a 100644 --- a/web/scan_terminal_city_row.py +++ b/web/scan_terminal_city_row.py @@ -8,6 +8,10 @@ from typing import Any, Dict, List, Optional from src.analysis.deb_algorithm import calculate_deb_prediction, calculate_dynamic_weights from src.analysis.trend_engine import calculate_prob_distribution +from src.data_collection.nws_open_meteo_sources import ( + OPEN_METEO_MULTI_MODEL_ORDER, + _parse_open_meteo_multi_model_daily, +) from src.data_collection.multi_model_freshness import multi_model_forecasts_for_local_date from src.database.db_manager import DBManager from src.utils.refresh_policy import SCAN_ROWS_REFRESH_SEC @@ -22,6 +26,7 @@ from web.services.city_payloads import aggregate_runway_history SCAN_ROW_RUNWAY_HISTORY_RESOLUTION = "10m" SCAN_ROW_MAX_RUNWAY_POINTS = 144 +SCAN_TERMINAL_MULTI_MODEL_BATCH_SIZE = 20 _PANEL_CACHE_DB = DBManager() _analyze = None # compatibility hook for tests that assert scan terminal stays cache-only. SCAN_PANEL_CACHE_MAX_AGE_SEC = max(300, int(SCAN_ROWS_REFRESH_SEC) * 3) @@ -101,9 +106,192 @@ def _model_spread_sigma(forecasts: Dict[str, float], temp_symbol: str) -> float: return max(0.8 * scale, min(4.0 * scale, spread / 2.0)) +def _multi_model_cache_key( + city: str, + lat: float, + lon: float, + *, + use_fahrenheit: bool, +) -> str: + version = str(getattr(_weather, "multi_model_cache_version", "v5") or "v5") + return ( + f"{round(float(lat), 4)}:{round(float(lon), 4)}:{str(city or '').strip().lower()}:" + f"{'f' if use_fahrenheit else 'c'}:{version}" + ) + + +def _read_cached_multi_model_for_today( + city: str, + *, + lat: float, + lon: float, + use_fahrenheit: bool, + local_date: str, +) -> Optional[Dict[str, Any]]: + maybe_reload = getattr(_weather, "_maybe_reload_open_meteo_disk_cache", None) + if callable(maybe_reload): + try: + maybe_reload() + except Exception: + pass + cache = getattr(_weather, "_multi_model_cache", None) + lock = getattr(_weather, "_multi_model_cache_lock", None) + if not isinstance(cache, dict) or lock is None: + return None + key = _multi_model_cache_key(city, lat, lon, use_fahrenheit=use_fahrenheit) + try: + with lock: + entry = cache.get(key) + data = entry.get("data") if isinstance(entry, dict) else None + except Exception: + return None + if isinstance(data, dict) and multi_model_forecasts_for_local_date(data, local_date): + return dict(data) + return None + + +def _store_multi_model_cache( + city: str, + payload: Dict[str, Any], + *, + lat: float, + lon: float, + use_fahrenheit: bool, +) -> None: + cache = getattr(_weather, "_multi_model_cache", None) + lock = getattr(_weather, "_multi_model_cache_lock", None) + if not isinstance(cache, dict) or lock is None: + return + key = _multi_model_cache_key(city, lat, lon, use_fahrenheit=use_fahrenheit) + try: + with lock: + cache[key] = {"t": time.time(), "data": dict(payload)} + except Exception: + return + + +def _fetch_scan_terminal_multi_model_batch(city_names: List[str]) -> Dict[str, Dict[str, Any]]: + """Fetch daily max multi-model payloads for scan rows in batch. + + The scan terminal only needs today's max per model. A batched daily request + avoids 50 per-city calls while still preserving city-local dates. + """ + grouped: Dict[bool, List[Dict[str, Any]]] = {False: [], True: []} + results: Dict[str, Dict[str, Any]] = {} + for city in city_names: + city_meta = CITIES.get(city) or {} + lat = _safe_float(city_meta.get("lat")) + lon = _safe_float(city_meta.get("lon")) + if lat is None or lon is None: + continue + use_fahrenheit = bool(city_meta.get("f")) + local_date = _city_local_date(city, _safe_int(city_meta.get("tz"), 0)) + cached = _read_cached_multi_model_for_today( + city, + lat=lat, + lon=lon, + use_fahrenheit=use_fahrenheit, + local_date=local_date, + ) + if cached: + results[city] = cached + continue + grouped[use_fahrenheit].append( + { + "city": city, + "lat": lat, + "lon": lon, + "local_date": local_date, + } + ) + + http_get = getattr(_weather, "_http_get", None) + if not callable(http_get): + return results + + stored_any = False + for use_fahrenheit, unit_items in grouped.items(): + if not unit_items: + continue + for start in range(0, len(unit_items), SCAN_TERMINAL_MULTI_MODEL_BATCH_SIZE): + items = unit_items[start : start + SCAN_TERMINAL_MULTI_MODEL_BATCH_SIZE] + if not items: + continue + try: + wait_slot = getattr(_weather, "_wait_open_meteo_slot", None) + if callable(wait_slot): + wait_slot("scan-terminal-multi-model-batch") + params: Dict[str, Any] = { + "latitude": ",".join(str(item["lat"]) for item in items), + "longitude": ",".join(str(item["lon"]) for item in items), + "daily": "temperature_2m_max", + "models": ",".join(OPEN_METEO_MULTI_MODEL_ORDER), + "timezone": "auto", + "forecast_days": 3, + } + if use_fahrenheit: + params["temperature_unit"] = "fahrenheit" + response = http_get( + "https://api.open-meteo.com/v1/forecast", + params=params, + timeout=max(10.0, float(getattr(_weather, "open_meteo_timeout_sec", 5.0))), + ) + response.raise_for_status() + raw = response.json() + payloads = raw if isinstance(raw, list) else [raw] + for item, location_payload in zip(items, payloads): + daily = location_payload.get("daily", {}) if isinstance(location_payload, dict) else {} + if not isinstance(daily, dict): + continue + dates, daily_forecasts, model_metadata, model_keys = _parse_open_meteo_multi_model_daily(daily) + if not daily_forecasts: + continue + local_date = item["local_date"] + forecasts = daily_forecasts.get(local_date) or {} + if not forecasts: + continue + city = str(item["city"]) + result = { + "source": "multi_model", + "provider": "open-meteo", + "forecasts": forecasts, + "daily_forecasts": daily_forecasts, + "hourly_times": [], + "hourly_forecasts": {}, + "model_metadata": model_metadata, + "model_keys": model_keys, + "dates": dates, + "unit": "fahrenheit" if use_fahrenheit else "celsius", + "attribution": "Open-Meteo forecast model API; underlying models from ECMWF, DWD, ECCC, NOAA/NCEP, Google and JMA.", + "scan_terminal_batch": True, + } + results[city] = result + _store_multi_model_cache( + city, + result, + lat=float(item["lat"]), + lon=float(item["lon"]), + use_fahrenheit=use_fahrenheit, + ) + stored_any = True + except Exception: + continue + if stored_any: + flush = getattr(_weather, "_flush_open_meteo_disk_cache", None) + if callable(flush): + try: + flush() + except Exception: + pass + return results + + def _fetch_today_forecast_panel_payload( city: str, payload: Dict[str, Any], + *, + multi_model_override: Optional[Dict[str, Any]] = None, + allow_direct_fetch: bool = True, ) -> Optional[Dict[str, Any]]: city_meta = CITIES.get(city) or {} lat = _safe_float(city_meta.get("lat")) @@ -120,15 +308,19 @@ def _fetch_today_forecast_panel_payload( local_date = _city_local_date(city, tz_offset_int) local_time = _city_local_time(city, tz_offset_int) - try: - multi_model = _weather.fetch_multi_model( - lat, - lon, - city=city, - use_fahrenheit=use_fahrenheit, - ) - except Exception: - multi_model = None + multi_model = multi_model_override + if not isinstance(multi_model, dict): + if not allow_direct_fetch: + return None + try: + multi_model = _weather.fetch_multi_model( + lat, + lon, + city=city, + use_fahrenheit=use_fahrenheit, + ) + except Exception: + multi_model = None forecasts = multi_model_forecasts_for_local_date(multi_model, local_date) if not forecasts: return None @@ -186,7 +378,34 @@ def _fetch_today_forecast_panel_payload( } -def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict[str, Any]]: +def _build_forecast_only_panel_payload( + city: str, + multi_model: Dict[str, Any], +) -> Optional[Dict[str, Any]]: + city_meta = CITIES.get(city) or {} + temp_symbol = "°F" if bool(city_meta.get("f")) else "°C" + return _fetch_today_forecast_panel_payload( + city, + { + "display_name": city_meta.get("name") or city_meta.get("display_name") or city, + "current": {}, + "risk": {}, + "probabilities": {}, + "temp_symbol": temp_symbol, + "utc_offset_seconds": _safe_int(city_meta.get("tz"), 0), + }, + multi_model_override=multi_model, + allow_direct_fetch=False, + ) + + +def _load_scan_panel_payload( + city: str, + *, + force_refresh: bool, + multi_model_override: Optional[Dict[str, Any]] = None, + allow_direct_fetch: bool = True, +) -> Optional[Dict[str, Any]]: refresh_already_queued = False cached_entry = _PANEL_CACHE_DB.get_city_cache("panel", city) cached_payload = cached_entry.get("payload") if isinstance(cached_entry, dict) else None @@ -196,7 +415,12 @@ def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict return cached_payload _enqueue_scan_terminal_refresh(city, reason=stale_reason or "scan_terminal_force_forecast_refresh") refresh_already_queued = True - refreshed_payload = _fetch_today_forecast_panel_payload(city, cached_payload) + refreshed_payload = _fetch_today_forecast_panel_payload( + city, + cached_payload, + multi_model_override=multi_model_override, + allow_direct_fetch=allow_direct_fetch, + ) if refreshed_payload: return refreshed_payload @@ -212,7 +436,12 @@ def _load_scan_panel_payload(city: str, *, force_refresh: bool) -> Optional[Dict city_meta = CITIES.get(city) or {} payload.setdefault("display_name", city_meta.get("name") or city_meta.get("display_name") or city) payload.setdefault("temp_symbol", canonical.get("temp_symbol") or "°C") - refreshed_payload = _fetch_today_forecast_panel_payload(city, payload) + refreshed_payload = _fetch_today_forecast_panel_payload( + city, + payload, + multi_model_override=multi_model_override, + allow_direct_fetch=allow_direct_fetch, + ) if refreshed_payload: payload = refreshed_payload _enqueue_scan_terminal_refresh(city, reason="scan_terminal_canonical_fallback") @@ -353,8 +582,16 @@ def _scan_city_terminal_rows( filters: Dict[str, Any], *, force_refresh: bool = False, + multi_model_override: Optional[Dict[str, Any]] = None, + allow_direct_fetch: bool = True, ) -> Dict[str, Any]: - return _scan_city_terminal_rows_quick(city, filters, force_refresh=force_refresh) + return _scan_city_terminal_rows_quick( + city, + filters, + force_refresh=force_refresh, + multi_model_override=multi_model_override, + allow_direct_fetch=allow_direct_fetch, + ) def _scan_city_terminal_rows_quick( @@ -362,10 +599,28 @@ def _scan_city_terminal_rows_quick( filters: Dict[str, Any], *, force_refresh: bool = False, + multi_model_override: Optional[Dict[str, Any]] = None, + allow_direct_fetch: bool = True, ) -> Dict[str, Any]: """Fast path that returns cached analysis rows only — returns a single row per city with cached analysis data (Obs, DEB, probabilities) but no market prices.""" - data = _load_scan_panel_payload(city, force_refresh=force_refresh) + if isinstance(multi_model_override, dict) and multi_model_override: + data = _build_forecast_only_panel_payload(city, multi_model_override) + if data: + row = _build_quick_row(city=city, data=data) + return { + "city": city, + "rows": [row] if row else [], + "candidate_total": 1, + "primary_scores": [float(row.get("final_score") or 0)] if row else [], + } + + data = _load_scan_panel_payload( + city, + force_refresh=force_refresh, + multi_model_override=multi_model_override, + allow_direct_fetch=allow_direct_fetch, + ) if not data: return { "city": city, diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index dcff5f4b..3f9bdee2 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -29,7 +29,10 @@ from web.scan_terminal_cache import ( set_cached_scan_terminal_payload, set_scan_terminal_failure_state, ) -from web.scan_terminal_city_row import _scan_city_terminal_rows +from web.scan_terminal_city_row import ( + _fetch_scan_terminal_multi_model_batch, + _scan_city_terminal_rows, +) from web.scan_terminal_filters import ( normalize_scan_terminal_filters as _normalize_scan_terminal_filters, ) @@ -67,6 +70,19 @@ def _rows_count(payload: Dict[str, Any]) -> int: return len(rows) if isinstance(rows, list) else 0 +def _model_bearing_rows_count(rows: Any) -> int: + if not isinstance(rows, list): + return 0 + count = 0 + for row in rows: + if not isinstance(row, dict): + continue + sources = row.get("model_cluster_sources") + if isinstance(sources, dict) and sources: + count += 1 + return count + + def _success_payload_within_stale_window(cached_entry: Dict[str, Any]) -> bool: timestamp = cached_entry.get("success_t") or cached_entry.get("t") try: @@ -116,6 +132,10 @@ def _build_stale_payload_for_timeout_if_better_cached( success_payload = cached_entry.get("success_payload") if not isinstance(success_payload, dict) or not success_payload.get("rows"): return None + if _model_bearing_rows_count(ranked_rows) > _model_bearing_rows_count( + success_payload.get("rows") + ): + return None if _rows_count(success_payload) < len(ranked_rows): return None @@ -184,59 +204,86 @@ def _build_scan_terminal_payload_uncached( return _tz_region(tz)["key"] == region_filter city_names = [c for c in city_names if _city_in_region(c)] - max_workers = max(1, min(SCAN_TERMINAL_MAX_WORKERS, len(city_names))) city_results: List[Dict[str, Any]] = [] failed_cities: List[str] = [] failed_reasons: List[str] = [] + multi_model_overrides = _fetch_scan_terminal_multi_model_batch(city_names) + scan_city_names = ( + [city_name for city_name in city_names if city_name in multi_model_overrides] + if multi_model_overrides + else city_names + ) + max_workers = max(1, min(SCAN_TERMINAL_MAX_WORKERS, len(scan_city_names))) timed_out = False timeout_message: Optional[str] = None - executor = ThreadPoolExecutor(max_workers=max_workers) - future_map = { - executor.submit( - _scan_city_terminal_rows, - city_name, - filters, - force_refresh=force_refresh, - ): city_name - for city_name in city_names - } - try: - try: - completed = as_completed( - future_map, - timeout=build_timeout_sec, - ) - for future in completed: - city_name = future_map[future] - try: - city_results.append(future.result()) - except Exception as exc: - failed_cities.append(city_name) - failed_reasons.append(str(exc)) - logger.warning( - "scan terminal city failed city={}: {}", city_name, exc + if multi_model_overrides: + for city_name in scan_city_names: + try: + city_results.append( + _scan_city_terminal_rows( + city_name, + filters, + force_refresh=force_refresh, + multi_model_override=multi_model_overrides.get(city_name), + allow_direct_fetch=False, ) - except FutureTimeoutError: - timed_out = True - timeout_message = ( - f"scan terminal build timed out after {build_timeout_sec:g}s" - ) - failed_reasons.append(timeout_message) - for future, city_name in future_map.items(): - if not future.done(): - future.cancel() - failed_cities.append(city_name) - logger.warning( - "{}; completed={}/{}", - timeout_message, - len(city_results), - len(city_names), - ) - finally: - executor.shutdown(wait=False) + ) + except Exception as exc: + failed_cities.append(city_name) + failed_reasons.append(str(exc)) + logger.warning( + "scan terminal city failed city={}: {}", city_name, exc + ) + else: + executor = ThreadPoolExecutor(max_workers=max_workers) + future_map = { + executor.submit( + _scan_city_terminal_rows, + city_name, + filters, + force_refresh=force_refresh, + multi_model_override=multi_model_overrides.get(city_name), + allow_direct_fetch=False, + ): city_name + for city_name in scan_city_names + } + try: + try: + completed = as_completed( + future_map, + timeout=build_timeout_sec, + ) + for future in completed: + city_name = future_map[future] + try: + city_results.append(future.result()) + except Exception as exc: + failed_cities.append(city_name) + failed_reasons.append(str(exc)) + logger.warning( + "scan terminal city failed city={}: {}", city_name, exc + ) + except FutureTimeoutError: + timed_out = True + timeout_message = ( + f"scan terminal build timed out after {build_timeout_sec:g}s" + ) + failed_reasons.append(timeout_message) + for future, city_name in future_map.items(): + if not future.done(): + future.cancel() + failed_cities.append(city_name) + logger.warning( + "{}; completed={}/{}", + timeout_message, + len(city_results), + len(scan_city_names), + ) + finally: + executor.shutdown(wait=False) - if city_names and len(failed_cities) >= len(city_names): + if scan_city_names and len(failed_cities) >= len(scan_city_names): error_message = ( failed_reasons[0] if failed_reasons else "all city market scans failed" )