From 6b7a4575ad5b4462786ba53c77e177257b340539 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 25 May 2026 17:51:10 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8B=86=E5=88=86=20analysis=5Fservice?= =?UTF-8?q?=EF=BC=9A=E5=89=A9=E4=BD=99=E5=B7=A5=E5=85=B7=E5=87=BD=E6=95=B0?= =?UTF-8?q?=E5=85=A8=E9=83=A8=E7=A7=BB=E8=87=B3=20analysis=5Futils.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 提取 _mgm_hourly_high/_dedupe_forecast_daily/_format_observation_time_local/_parse_local_hour/_parse_utc_datetime/_metar_is_current_local_day/_is_plausible_city_temp。统一 parse_utc_datetime 副本。analysis_service 2082→1966 行。 --- frontend/app/api/scan/terminal/route.ts | 1 + .../dashboard/ScanTerminalDashboard.tsx | 11 +- .../scan-terminal/scan-terminal-client.ts | 5 + .../scan-terminal/use-scan-terminal-query.ts | 7 +- tests/test_scan_terminal_modules.py | 2 + web/analysis_service.py | 130 +----------------- web/routers/scan.py | 2 + web/scan_terminal_filters.py | 2 + web/scan_terminal_service.py | 8 ++ web/services/analysis_utils.py | 112 +++++++++++++++ web/services/observation_freshness.py | 14 +- web/services/scan_api.py | 3 + 12 files changed, 158 insertions(+), 139 deletions(-) diff --git a/frontend/app/api/scan/terminal/route.ts b/frontend/app/api/scan/terminal/route.ts index 8910c7be..2d6ec539 100644 --- a/frontend/app/api/scan/terminal/route.ts +++ b/frontend/app/api/scan/terminal/route.ts @@ -31,6 +31,7 @@ export async function GET(req: NextRequest) { "limit", "force_refresh", "skip_polymarket", + "timezone_offset_seconds", ]) { const value = req.nextUrl.searchParams.get(key); if (value != null && value !== "") { diff --git a/frontend/components/dashboard/ScanTerminalDashboard.tsx b/frontend/components/dashboard/ScanTerminalDashboard.tsx index f2ef4d44..f751de14 100644 --- a/frontend/components/dashboard/ScanTerminalDashboard.tsx +++ b/frontend/components/dashboard/ScanTerminalDashboard.tsx @@ -672,6 +672,8 @@ function ScanTerminalScreen() { const userLocalTime = useUserLocalClock(); const { themeMode } = useScanTerminalTheme(); const [selectedRegionKey, setSelectedRegionKey] = useState("east_asia"); + const [localTimezoneOffsetSeconds, setLocalTimezoneOffsetSeconds] = useState(null); + const [useLocalTimezoneDefault, setUseLocalTimezoneDefault] = useState(true); useEffect(() => { let cancelled = false; @@ -743,12 +745,19 @@ function ScanTerminalScreen() { useEffect(() => { setSelectedRegionKey(detectLocalRegion()); + setLocalTimezoneOffsetSeconds(-new Date().getTimezoneOffset() * 60); + }, []); + + const selectRegionManually = useCallback((key: string) => { + setUseLocalTimezoneDefault(false); + setSelectedRegionKey(key); }, []); const { refreshScanTerminalManually, scanLoading, terminalData } = useScanTerminalQuery({ isPro, proAccessLoading: !hydrated || (proAccess.loading && !canUseLocalFullAccess), + timezoneOffsetSeconds: useLocalTimezoneDefault ? localTimezoneOffsetSeconds : null, tradingRegion: selectedRegionKey, }); const rows = useMemo( @@ -831,7 +840,7 @@ function ScanTerminalScreen() { selectedCity={selectedCity} setSelectedCity={setSelectedCity} selectedRegionKey={selectedRegionKey} - setSelectedRegionKey={setSelectedRegionKey} + setSelectedRegionKey={selectRegionManually} /> ); } diff --git a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts index 607f727d..9e0b9eb0 100644 --- a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts +++ b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts @@ -53,6 +53,7 @@ type AiCityStreamEvent = { type TerminalQueryOptions = { forceRefresh?: boolean; signal?: AbortSignal; + timezoneOffsetSeconds?: number | null; tradingRegion?: string; }; @@ -254,6 +255,7 @@ async function readAiCityForecastStream( async function getTerminal({ forceRefresh = false, signal, + timezoneOffsetSeconds, tradingRegion, }: TerminalQueryOptions = {}) { const params = new URLSearchParams({ @@ -271,6 +273,9 @@ async function getTerminal({ if (tradingRegion && tradingRegion !== "all") { params.set("trading_region", tradingRegion); } + if (Number.isFinite(timezoneOffsetSeconds)) { + params.set("timezone_offset_seconds", String(Math.trunc(Number(timezoneOffsetSeconds)))); + } if (forceRefresh) { params.set("_ts", String(Date.now())); } diff --git a/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts b/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts index e1a41279..71be087f 100644 --- a/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts +++ b/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts @@ -13,10 +13,12 @@ import type { ScanTerminalResponse } from "@/lib/dashboard-types"; export function useScanTerminalQuery({ isPro, proAccessLoading, + timezoneOffsetSeconds, tradingRegion, }: { isPro: boolean; proAccessLoading: boolean; + timezoneOffsetSeconds?: number | null; tradingRegion?: string; }) { const { @@ -50,12 +52,13 @@ export function useScanTerminalQuery({ scanTerminalClient.getTerminal({ forceRefresh, signal, + timezoneOffsetSeconds, tradingRegion, }), showLoading, }); }, - [isPro, proAccessLoading, run, tradingRegion], + [isPro, proAccessLoading, run, timezoneOffsetSeconds, tradingRegion], ); useEffect(() => { @@ -65,7 +68,7 @@ export function useScanTerminalQuery({ return; } void fetchScanTerminal({ forceRefresh: false, showLoading: true }); - }, [fetchScanTerminal, isPro, proAccessLoading, reset, tradingRegion]); + }, [fetchScanTerminal, isPro, proAccessLoading, reset, timezoneOffsetSeconds, tradingRegion]); const refreshScanTerminalManually = useCallback(() => { if ( diff --git a/tests/test_scan_terminal_modules.py b/tests/test_scan_terminal_modules.py index 737a278a..bbf40d21 100644 --- a/tests/test_scan_terminal_modules.py +++ b/tests/test_scan_terminal_modules.py @@ -17,6 +17,7 @@ def test_normalize_scan_terminal_filters_clamps_and_swaps_bounds(): "limit": 999, "high_liquidity_only": True, "min_liquidity": 100, + "timezone_offset_seconds": "28800", } ) @@ -24,6 +25,7 @@ def test_normalize_scan_terminal_filters_clamps_and_swaps_bounds(): assert filters["max_price"] == 1.0 assert filters["limit"] == 200 assert filters["min_liquidity"] == 5000.0 + assert filters["timezone_offset_seconds"] == 28800 def test_ranked_scan_terminal_result_sorts_and_summarizes_unique_markets(): diff --git a/web/analysis_service.py b/web/analysis_service.py index 6c0289f2..5728e8d0 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -1,6 +1,5 @@ from __future__ import annotations -import re import time as _time import threading from concurrent.futures import ThreadPoolExecutor @@ -43,8 +42,15 @@ from web.services.analysis_utils import ( add_signal as _add_signal, bucket_label as _bucket_label, bucket_label_from_value as _bucket_label_from_value, + dedupe_forecast_daily as _dedupe_forecast_daily, format_clock_minutes as _format_clock_minutes, + format_observation_time_local as _format_observation_time_local, + is_plausible_city_temp as _is_plausible_city_temp, + metar_is_current_local_day as _metar_is_current_local_day, + mgm_hourly_high as _mgm_hourly_high, next_observation_clock as _next_observation_clock, + parse_local_hour as _parse_local_hour, + parse_utc_datetime as _parse_utc_datetime, top_probability_bucket as _top_probability_bucket, ) from web.services.analysis_signals import ( @@ -79,128 +85,6 @@ HIGH_FREQ_AIRPORT_ANALYSIS_CITIES = { "wuhan", } - -def _mgm_hourly_high(mgm: Dict[str, Any]) -> Optional[float]: - hourly = mgm.get("hourly") if isinstance(mgm, dict) else [] - if not isinstance(hourly, list): - return None - values = [] - for row in hourly: - if not isinstance(row, dict): - continue - value = _sf(row.get("temp")) - if value is not None: - values.append(value) - return max(values) if values else None -_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, -} -_SUMMARY_CACHE_LOCK = threading.Lock() -_SUMMARY_CACHE_MAXSIZE = 128 -_SUMMARY_CACHE = LRUDict(maxsize=_SUMMARY_CACHE_MAXSIZE) -def _dedupe_forecast_daily(rows: Any) -> list[Dict[str, Any]]: - if not isinstance(rows, list): - return [] - seen = set() - out = [] - for row in rows: - if not isinstance(row, dict): - continue - date = str(row.get("date") or "").strip() - if not date or date in seen: - continue - seen.add(date) - out.append(row) - return out - - -def _format_observation_time_local(value: Any, utc_offset: int) -> str: - raw = str(value or "").strip() - if not raw: - return "" - if "T" in raw: - try: - dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone(timedelta(seconds=utc_offset))).strftime("%H:%M") - except Exception: - pass - match = re.search(r"(\d{1,2}):(\d{2})", raw) - if match: - return f"{int(match.group(1)):02d}:{match.group(2)}" - return raw[:16] - - -def _fetch_nmc_current_fallback(city: str, *, use_fahrenheit: bool) -> Dict[str, Any]: - return {} - - -def _is_plausible_city_temp(city: str, value: Any, unit: str = "°C") -> bool: - temp = _sf(value) - if temp is None: - return False - meta = CITY_REGISTRY.get(str(city or "").strip().lower(), {}) or {} - min_c = _sf(meta.get("min_plausible_metar_temp_c")) - if min_c is None: - return True - min_value = min_c * 9 / 5 + 32 if str(unit or "").upper().endswith("F") else min_c - return temp >= min_value - - -def _parse_local_hour(local_time_str: Optional[str]) -> Optional[int]: - if not local_time_str: - return None - try: - parts = str(local_time_str).strip().split(":") - hour = int(parts[0]) - if 0 <= hour <= 23: - return hour - except Exception: - pass - return None - - -def _parse_utc_datetime(value: Any) -> Optional[datetime]: - raw = str(value or "").strip() - if not raw or "T" not in raw: - return None - try: - dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) - except Exception: - return None - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc) - - -def _metar_is_current_local_day( - metar: Dict[str, Any], - *, - local_date: str, - utc_offset: int, -) -> bool: - if not isinstance(metar, dict) or not metar: - return False - if metar.get("stale_for_today") is True: - return False - observation_local_date = str(metar.get("observation_local_date") or "").strip() - if observation_local_date: - return observation_local_date == local_date - obs_dt = _parse_utc_datetime(metar.get("observation_time")) - if obs_dt is None: - return True - local_dt = obs_dt.astimezone(timezone(timedelta(seconds=utc_offset))) - return local_dt.strftime("%Y-%m-%d") == local_date - - def _record_analysis_cache_event(*, city: str, hit: bool, force_refresh: bool) -> None: now = datetime.now(timezone.utc).isoformat() with _ANALYSIS_CACHE_STATS_LOCK: diff --git a/web/routers/scan.py b/web/routers/scan.py index 15fe8cc8..48ee97b8 100644 --- a/web/routers/scan.py +++ b/web/routers/scan.py @@ -29,6 +29,7 @@ async def scan_terminal( region: str = "", trading_region: str = "", skip_polymarket: bool = False, + timezone_offset_seconds: int | None = None, ): return await get_scan_terminal_payload( request, @@ -44,6 +45,7 @@ async def scan_terminal( force_refresh=force_refresh, region=region or trading_region or None, skip_polymarket=skip_polymarket, + timezone_offset_seconds=timezone_offset_seconds, ) diff --git a/web/scan_terminal_filters.py b/web/scan_terminal_filters.py index a51600d7..b66c7640 100644 --- a/web/scan_terminal_filters.py +++ b/web/scan_terminal_filters.py @@ -54,6 +54,8 @@ def normalize_scan_terminal_filters( trading_region = str(raw.get("trading_region") or "").strip().lower() if trading_region and trading_region not in ("all", ""): result["trading_region"] = trading_region + if raw.get("timezone_offset_seconds") is not None: + result["timezone_offset_seconds"] = safe_int(raw.get("timezone_offset_seconds"), 0) return result diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index 87b2057e..e8d5e673 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -1111,6 +1111,14 @@ def _build_scan_terminal_payload_uncached( try: city_names = list(CITIES.keys()) + timezone_offset = filters.get("timezone_offset_seconds") + if timezone_offset is not None: + target_tz = int(timezone_offset) + city_names = [ + city_name + for city_name in city_names + if int((CITIES.get(city_name) or {}).get("tz", 0)) == target_tz + ] region_filter = str(filters.get("trading_region") or "").strip().lower() if region_filter and region_filter not in ("all", ""): from web.scan_terminal_filters import market_region_from_tz_offset as _tz_region diff --git a/web/services/analysis_utils.py b/web/services/analysis_utils.py index 4fba7e0c..7fe0bb0a 100644 --- a/web/services/analysis_utils.py +++ b/web/services/analysis_utils.py @@ -6,6 +6,7 @@ Pure helpers: clock arithmetic, bucket labelling, signal packaging. from __future__ import annotations import re +from datetime import datetime, timezone, timedelta from typing import Any, Dict, Optional from web.core import _sf @@ -92,3 +93,114 @@ def add_signal( "summary_en": summary_en or summary, } ) + + +# ── Time / date helpers ──────────────────────────────────────────────── + +def parse_utc_datetime(value: Any) -> Optional[datetime]: + raw = str(value or "").strip() + if not raw or "T" not in raw: + return None + try: + dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except Exception: + return None + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + + +def format_observation_time_local(value: Any, utc_offset: int) -> str: + raw = str(value or "").strip() + if not raw: + return "" + if "T" in raw: + try: + dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone(timedelta(seconds=utc_offset))).strftime("%H:%M") + except Exception: + pass + import re + match = re.search(r"(\d{1,2}):(\d{2})", raw) + if match: + return f"{int(match.group(1)):02d}:{match.group(2)}" + return raw[:16] + + +def parse_local_hour(local_time_str: Optional[str]) -> Optional[int]: + if not local_time_str: + return None + try: + parts = str(local_time_str).strip().split(":") + hour = int(parts[0]) + if 0 <= hour <= 23: + return hour + except Exception: + pass + return None + + +def metar_is_current_local_day( + metar: Dict[str, Any], + *, + local_date: str, + utc_offset: int, +) -> bool: + if not isinstance(metar, dict) or not metar: + return False + if metar.get("stale_for_today") is True: + return False + observation_local_date = str(metar.get("observation_local_date") or "").strip() + if observation_local_date: + return observation_local_date == local_date + obs_dt = parse_utc_datetime(metar.get("observation_time")) + if obs_dt is None: + return True + local_dt = obs_dt.astimezone(timezone(timedelta(seconds=utc_offset))) + return local_dt.strftime("%Y-%m-%d") == local_date + + +def is_plausible_city_temp(city: str, value: Any, unit: str = "°C") -> bool: + from src.data_collection.city_registry import CITY_REGISTRY + + temp = _sf(value) + if temp is None: + return False + meta = CITY_REGISTRY.get(str(city or "").strip().lower(), {}) or {} + min_c = _sf(meta.get("min_plausible_metar_temp_c")) + if min_c is None: + return True + min_value = min_c * 9 / 5 + 32 if str(unit or "").upper().endswith("F") else min_c + return temp >= min_value + + +def dedupe_forecast_daily(rows: Any) -> list: + if not isinstance(rows, list): + return [] + seen = set() + out = [] + for row in rows: + if not isinstance(row, dict): + continue + date = str(row.get("date") or "").strip() + if not date or date in seen: + continue + seen.add(date) + out.append(row) + return out + + +def mgm_hourly_high(mgm: Dict[str, Any]) -> Optional[float]: + hourly = mgm.get("hourly") if isinstance(mgm, dict) else [] + if not isinstance(hourly, list): + return None + values = [] + for row in hourly: + if not isinstance(row, dict): + continue + value = _sf(row.get("temp")) + if value is not None: + values.append(value) + return max(values) if values else None diff --git a/web/services/observation_freshness.py b/web/services/observation_freshness.py index f3fd0173..01516062 100644 --- a/web/services/observation_freshness.py +++ b/web/services/observation_freshness.py @@ -8,6 +8,7 @@ from __future__ import annotations from datetime import datetime, timezone, timedelta from typing import Any, Dict, Optional +from web.services.analysis_utils import parse_utc_datetime _OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = { "amos": { @@ -90,19 +91,6 @@ _OBSERVATION_SOURCE_PROFILES: Dict[str, Dict[str, Any]] = { } -def parse_utc_datetime(value: Any) -> Optional[datetime]: - raw = str(value or "").strip() - if not raw or "T" not in raw: - return None - try: - dt = datetime.fromisoformat(raw.replace("Z", "+00:00")) - except Exception: - return None - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc) - - def observation_age_min(value: Any, now_utc: Optional[datetime] = None) -> Optional[int]: obs_dt = parse_utc_datetime(value) if obs_dt is None: diff --git a/web/services/scan_api.py b/web/services/scan_api.py index 917ea8bc..4fe3f4dd 100644 --- a/web/services/scan_api.py +++ b/web/services/scan_api.py @@ -48,6 +48,7 @@ async def get_scan_terminal_payload( force_refresh: bool = False, region: str = "", skip_polymarket: bool = False, + timezone_offset_seconds: int | None = None, ) -> Dict[str, Any]: legacy_routes._assert_entitlement(request) filters: Dict[str, Any] = { @@ -62,6 +63,8 @@ async def get_scan_terminal_payload( "limit": limit, "skip_polymarket": skip_polymarket, } + if timezone_offset_seconds is not None: + filters["timezone_offset_seconds"] = timezone_offset_seconds if region: filters["trading_region"] = region return await run_in_threadpool(