From 40564b545d63836fd18cfc2720ebb6a008b44bfd Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 22 Apr 2026 04:54:19 +0800 Subject: [PATCH] Optimize probability hub market refresh --- .../probability-hub/ProbabilityHubPage.tsx | 200 +++++++++++++----- frontend/lib/dashboard-client.ts | 35 ++- src/data_collection/polymarket_readonly.py | 4 +- web/routes.py | 145 ++++++++++++- 4 files changed, 326 insertions(+), 58 deletions(-) diff --git a/frontend/components/probability-hub/ProbabilityHubPage.tsx b/frontend/components/probability-hub/ProbabilityHubPage.tsx index 9907be28..6880583d 100644 --- a/frontend/components/probability-hub/ProbabilityHubPage.tsx +++ b/frontend/components/probability-hub/ProbabilityHubPage.tsx @@ -1,6 +1,6 @@ "use client"; -import { useCallback, useEffect, useMemo, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import clsx from "clsx"; import dashboardStyles from "@/components/dashboard/Dashboard.module.css"; import { HeaderBar } from "@/components/dashboard/HeaderBar"; @@ -13,7 +13,7 @@ import styles from "./ProbabilityHubPage.module.css"; const DETAIL_BATCH_SIZE = 6; const FULL_REFRESH_INTERVAL_MS = 60_000; -const MARKET_REFRESH_INTERVAL_MS = 5_000; +const MARKET_REFRESH_INTERVAL_MS = 20_000; type FilterMode = "all" | "market" | "high-risk"; type SortMode = "risk" | "edge" | "probability" | "updated"; @@ -112,6 +112,20 @@ function ProbabilityHubScreen() { const [lastMarketUpdatedAt, setLastMarketUpdatedAt] = useState(null); const [filterMode, setFilterMode] = useState("all"); const [sortMode, setSortMode] = useState("risk"); + const [visibleCityNames, setVisibleCityNames] = useState>(new Set()); + const cardNodesRef = useRef(new Map()); + const initialMarketRequestsRef = useRef(new Set()); + const marketRefreshInFlightRef = useRef(false); + const detailsByNameRef = useRef(detailsByName); + const citiesRef = useRef(cities); + + useEffect(() => { + detailsByNameRef.current = detailsByName; + }, [detailsByName]); + + useEffect(() => { + citiesRef.current = cities; + }, [cities]); const fetchCityDetails = useCallback( async (cityList: CityListItem[], force: boolean) => { @@ -122,7 +136,7 @@ function ProbabilityHubScreen() { const results = await Promise.allSettled( batch.map((city) => dashboardClient.getCityDetail(city.name, { - depth: "market", + depth: "panel", force, }), ), @@ -153,68 +167,76 @@ function ProbabilityHubScreen() { sourceDetails?: Record, sourceCities?: CityListItem[], ) => { - const detailMap = sourceDetails || detailsByName; + if (marketRefreshInFlightRef.current) return; + const detailMap = sourceDetails ? { ...sourceDetails } : { ...detailsByName }; const cityList = sourceCities || cities; const loadedCities = cityList.filter((city) => detailMap[city.name]); if (!loadedCities.length) return; let touched = false; + marketRefreshInFlightRef.current = true; - for (let index = 0; index < loadedCities.length; index += DETAIL_BATCH_SIZE) { - const batch = loadedCities.slice(index, index + DETAIL_BATCH_SIZE); - const results = await Promise.allSettled( - batch.map((city) => - dashboardClient.getCityMarketScan(city.name, { - force: false, - targetDate: detailMap[city.name]?.local_date || null, - marketSlug: detailMap[city.name]?.market_scan?.selected_slug || null, - }), - ), - ); + try { + for (let index = 0; index < loadedCities.length; index += DETAIL_BATCH_SIZE) { + const batch = loadedCities.slice(index, index + DETAIL_BATCH_SIZE); + const results = await Promise.allSettled( + batch.map((city) => + dashboardClient.getCityMarketScan(city.name, { + force: false, + targetDate: detailMap[city.name]?.local_date || null, + marketSlug: detailMap[city.name]?.market_scan?.selected_slug || null, + }), + ), + ); - const patch: Record = {}; - results.forEach((result, batchIndex) => { - if (result.status !== "fulfilled") return; - const city = batch[batchIndex]; - const previous = detailMap[city.name] || detailsByName[city.name]; - if (!previous) return; - const nextMarketScan = result.value.market_scan || previous.market_scan; - if (!nextMarketScan) return; - patch[city.name] = { - ...previous, - market_scan: nextMarketScan, - }; - }); + const patch: Record = {}; + results.forEach((result, batchIndex) => { + if (result.status !== "fulfilled") return; + const city = batch[batchIndex]; + const previous = detailMap[city.name] || detailsByName[city.name]; + if (!previous) return; + const nextMarketScan = result.value.market_scan || previous.market_scan; + if (!nextMarketScan) return; + patch[city.name] = { + ...previous, + market_scan: nextMarketScan, + }; + }); - if (Object.keys(patch).length) { - touched = true; - Object.assign(detailMap, patch); - setDetailsByName((previous) => ({ - ...previous, - ...patch, - })); + if (Object.keys(patch).length) { + touched = true; + Object.assign(detailMap, patch); + setDetailsByName((previous) => ({ + ...previous, + ...patch, + })); + } } - } - if (touched) { - setLastMarketUpdatedAt(new Date().toISOString()); + if (touched) { + setLastMarketUpdatedAt(new Date().toISOString()); + } + } finally { + marketRefreshInFlightRef.current = false; } }, [cities, detailsByName]); const loadAll = useCallback(async (force = false) => { setError(null); setRefreshing(true); - if (!cities.length || force) { + if (!citiesRef.current.length || force) { setLoading(true); } + if (force) { + initialMarketRequestsRef.current.clear(); + } try { const cityList = sortCities(await dashboardClient.getCities()); setCities(cityList); - const fetched = await fetchCityDetails(cityList, force); + await fetchCityDetails(cityList, force); setLastUpdatedAt(new Date().toISOString()); - await refreshMarketScans(fetched, cityList); } catch (loadError) { - if (!Object.keys(detailsByName).length && !cities.length) { + if (!Object.keys(detailsByNameRef.current).length && !citiesRef.current.length) { setError( loadError instanceof Error ? loadError.message @@ -229,7 +251,7 @@ function ProbabilityHubScreen() { setLoading(false); setRefreshing(false); } - }, [cities.length, detailsByName, fetchCityDetails, locale, refreshMarketScans]); + }, [fetchCityDetails, locale]); const retryMissingCities = useCallback(async () => { if (!cities.length) return; @@ -239,11 +261,14 @@ function ProbabilityHubScreen() { try { const fetched = await fetchCityDetails(missingCities, true); setLastUpdatedAt(new Date().toISOString()); - await refreshMarketScans(fetched, missingCities); + const visibleMissing = missingCities.filter((city) => visibleCityNames.has(city.name)); + if (visibleMissing.length) { + await refreshMarketScans(fetched, visibleMissing); + } } catch { // keep silent; page-level retry should not override the main error banner } - }, [cities, detailsByName, fetchCityDetails, refreshMarketScans]); + }, [cities, detailsByName, fetchCityDetails, refreshMarketScans, visibleCityNames]); useEffect(() => { void loadAll(false); @@ -270,11 +295,13 @@ function ProbabilityHubScreen() { useEffect(() => { const timer = window.setInterval(() => { if (document.visibilityState === "hidden") return; - void refreshMarketScans(); + const visibleCitiesForRefresh = cities.filter((city) => visibleCityNames.has(city.name)); + if (!visibleCitiesForRefresh.length) return; + void refreshMarketScans(undefined, visibleCitiesForRefresh); }, MARKET_REFRESH_INTERVAL_MS); return () => window.clearInterval(timer); - }, [refreshMarketScans]); + }, [cities, refreshMarketScans, visibleCityNames]); const loadedCount = Object.keys(detailsByName).length; const cityCount = cities.length; @@ -336,6 +363,71 @@ function ProbabilityHubScreen() { }); }, [cities, detailsByName, filterMode, sortMode]); + const visibleCitiesKey = useMemo( + () => visibleCities.map((city) => city.name).join("|"), + [visibleCities], + ); + + const setCardNode = useCallback((cityName: string, node: HTMLElement | null) => { + if (node) { + cardNodesRef.current.set(cityName, node); + } else { + cardNodesRef.current.delete(cityName); + } + }, []); + + useEffect(() => { + if (typeof IntersectionObserver === "undefined") { + setVisibleCityNames(new Set(visibleCities.slice(0, DETAIL_BATCH_SIZE).map((city) => city.name))); + return; + } + + const observer = new IntersectionObserver( + (entries) => { + setVisibleCityNames((previous) => { + const next = new Set(previous); + for (const entry of entries) { + const cityName = (entry.target as HTMLElement).dataset.cityName; + if (!cityName) continue; + if (entry.isIntersecting) { + next.add(cityName); + } else { + next.delete(cityName); + } + } + return next; + }); + }, + { + root: null, + rootMargin: "900px 0px", + threshold: 0.01, + }, + ); + + const namesInView = new Set(visibleCities.map((city) => city.name)); + for (const [cityName, node] of cardNodesRef.current.entries()) { + if (namesInView.has(cityName)) { + observer.observe(node); + } + } + + return () => observer.disconnect(); + }, [visibleCitiesKey, visibleCities]); + + useEffect(() => { + if (!visibleCityNames.size) return; + const citiesToWarm = cities.filter((city) => { + if (!visibleCityNames.has(city.name)) return false; + if (!detailsByName[city.name]) return false; + if (initialMarketRequestsRef.current.has(city.name)) return false; + return true; + }); + if (!citiesToWarm.length) return; + citiesToWarm.forEach((city) => initialMarketRequestsRef.current.add(city.name)); + void refreshMarketScans(undefined, citiesToWarm); + }, [cities, detailsByName, refreshMarketScans, visibleCityNames]); + return (
+
setCardNode(city.name, node)} + data-city-name={city.name} + className={styles.card} + >
{city.display_name}
@@ -502,7 +599,12 @@ function ProbabilityHubScreen() { } return ( -
+
setCardNode(city.name, node)} + data-city-name={city.name} + className={styles.card} + >
{detail.display_name}
diff --git a/frontend/lib/dashboard-client.ts b/frontend/lib/dashboard-client.ts index 71d6118c..d22094bd 100644 --- a/frontend/lib/dashboard-client.ts +++ b/frontend/lib/dashboard-client.ts @@ -13,6 +13,14 @@ const CACHE_TTL_MS = 5 * 60 * 1000; const pendingCityDetailRequests = new Map>(); const pendingHistoryRequests = new Map>(); const pendingCitySummaryRequests = new Map>(); +const pendingCityMarketScanRequests = new Map< + string, + Promise<{ + fetched_at?: string | null; + market_scan?: MarketScan | null; + selected_date?: string | null; + }> +>(); const PRIORITY_WARM_SESSION_KEY = "polyWeather_priority_warm_v1"; type CityCacheMeta = { @@ -247,8 +255,9 @@ export const dashboardClient = { targetDate?: string | null; }, ) { + const force = options?.force ?? false; const params = new URLSearchParams({ - force_refresh: String(options?.force ?? false), + force_refresh: String(force), _ts: String(Date.now()), }); if (options?.targetDate) { @@ -257,11 +266,31 @@ export const dashboardClient = { if (options?.marketSlug) { params.set("market_slug", options.marketSlug); } - return fetchJson<{ + const requestKey = [ + cityName, + force ? "force" : "cached", + options?.targetDate || "", + options?.marketSlug || "", + ].join("::"); + if (!force) { + const existing = pendingCityMarketScanRequests.get(requestKey); + if (existing) { + return existing; + } + } + const request = fetchJson<{ fetched_at?: string | null; market_scan?: MarketScan | null; selected_date?: string | null; - }>(`/api/city/${normalizeCityName(cityName)}/market-scan?${params.toString()}`); + }>(`/api/city/${normalizeCityName(cityName)}/market-scan?${params.toString()}`).finally( + () => { + pendingCityMarketScanRequests.delete(requestKey); + }, + ); + if (!force) { + pendingCityMarketScanRequests.set(requestKey, request); + } + return request; }, async getHistory(cityName: string, options?: { includeRecords?: boolean }) { diff --git a/src/data_collection/polymarket_readonly.py b/src/data_collection/polymarket_readonly.py index 94247788..8d312e86 100644 --- a/src/data_collection/polymarket_readonly.py +++ b/src/data_collection/polymarket_readonly.py @@ -605,7 +605,7 @@ class PolymarketReadOnlyLayer: ) all_bucket_limit = max( top_bucket_limit, - _safe_int(os.getenv("POLYMARKET_ALL_BUCKET_LIMIT", "24"), 24), + _safe_int(os.getenv("POLYMARKET_ALL_BUCKET_LIMIT", "8"), 8), ) all_buckets = self._build_top_temperature_buckets( city_key=city_key, @@ -613,8 +613,6 @@ class PolymarketReadOnlyLayer: primary_market=market, limit=all_bucket_limit, ) - if all_buckets: - self._hydrate_bucket_prices(all_buckets) top_buckets = list(all_buckets[:top_bucket_limit]) yes_payload = { diff --git a/web/routes.py b/web/routes.py index 714e974e..c89ccfe8 100644 --- a/web/routes.py +++ b/web/routes.py @@ -146,6 +146,10 @@ CITY_SUMMARY_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_SUMMARY_CAC CITY_PANEL_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_PANEL_CACHE_TTL_SEC", "300"))) CITY_NEARBY_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_NEARBY_CACHE_TTL_SEC", "480"))) CITY_MARKET_CACHE_TTL_SEC = max(30, int(os.getenv("POLYWEATHER_CITY_MARKET_CACHE_TTL_SEC", "900"))) +MARKET_SCAN_PAYLOAD_TTL_SEC = max( + 5, + int(os.getenv("POLYWEATHER_MARKET_SCAN_PAYLOAD_TTL_SEC", "20")), +) CITY_HISTORY_PREVIEW_CACHE_TTL_SEC = max( 60, int(os.getenv("POLYWEATHER_CITY_HISTORY_PREVIEW_CACHE_TTL_SEC", "1800")), @@ -162,6 +166,106 @@ def _city_cache_is_fresh(entry: Optional[dict], ttl_sec: int) -> bool: return (time.time() - updated_at_ts) < float(ttl_sec) +def _market_analysis_cache_is_fresh(entry: Optional[dict]) -> bool: + if not isinstance(entry, dict): + return False + payload = entry.get("payload") or {} + if isinstance(payload, dict): + cached_at_ts = float(payload.get("market_analysis_cached_at_ts") or 0.0) + if cached_at_ts > 0: + return (time.time() - cached_at_ts) < float(CITY_MARKET_CACHE_TTL_SEC) + return _city_cache_is_fresh(entry, CITY_MARKET_CACHE_TTL_SEC) + + +def _market_scan_cache_key( + data: dict, + market_slug: Optional[str] = None, + target_date: Optional[str] = None, +) -> str: + local_date = str(data.get("local_date") or "").strip() + requested_date = str(target_date or "").strip() + selected_date = requested_date or local_date + multi_model_daily = data.get("multi_model_daily") or {} + if requested_date and isinstance(multi_model_daily, dict) and requested_date not in multi_model_daily: + selected_date = local_date + normalized_slug = str(market_slug or "").strip().lower() + return f"{selected_date}|{normalized_slug}" + + +def _attach_market_scan_payload( + payload: dict, + *, + market_slug: Optional[str] = None, + target_date: Optional[str] = None, +) -> dict: + if not isinstance(payload, dict): + return payload + scan_payload = _build_city_market_scan_payload( + payload, + market_slug=market_slug, + target_date=target_date, + ) + now_ts = time.time() + payload["market_scan_payload"] = scan_payload + payload["market_scan_updated_at"] = datetime.now().isoformat() + payload["market_scan_updated_at_ts"] = now_ts + payload["market_scan_cache_key"] = _market_scan_cache_key( + payload, + market_slug=market_slug, + target_date=target_date, + ) + return payload + + +def _get_cached_market_scan_payload( + payload: dict, + *, + market_slug: Optional[str] = None, + target_date: Optional[str] = None, +) -> Optional[dict]: + if not isinstance(payload, dict): + return None + scan_payload = payload.get("market_scan_payload") + if not isinstance(scan_payload, dict): + return None + expected_key = _market_scan_cache_key( + payload, + market_slug=market_slug, + target_date=target_date, + ) + cached_key = str(payload.get("market_scan_cache_key") or "") + if cached_key != expected_key: + return None + updated_at_ts = float(payload.get("market_scan_updated_at_ts") or 0.0) + if updated_at_ts <= 0: + return None + if (time.time() - updated_at_ts) >= float(MARKET_SCAN_PAYLOAD_TTL_SEC): + return None + return scan_payload + + +def _refresh_market_scan_payload_from_cached_analysis( + city: str, + payload: dict, + *, + market_slug: Optional[str] = None, + target_date: Optional[str] = None, +) -> dict: + _attach_market_scan_payload( + payload, + market_slug=market_slug, + target_date=target_date, + ) + _CACHE_DB.set_city_cache( + "market", + city, + payload, + version="v1", + source_fingerprint=f"{city}:market", + ) + return payload.get("market_scan_payload") or {} + + def _refresh_city_summary_cache(city: str, force_refresh: bool = False) -> dict: data = _analyze_summary(city, force_refresh=force_refresh) payload = _build_city_summary_payload(data) @@ -201,6 +305,10 @@ def _refresh_city_nearby_cache(city: str, force_refresh: bool = False) -> dict: def _refresh_city_market_cache(city: str, force_refresh: bool = False) -> dict: payload = _analyze(city, force_refresh=force_refresh, include_llm_commentary=False, detail_mode="market") + now_ts = time.time() + payload["market_analysis_cached_at"] = datetime.now().isoformat() + payload["market_analysis_cached_at_ts"] = now_ts + _attach_market_scan_payload(payload) _CACHE_DB.set_city_cache( "market", city, @@ -865,7 +973,7 @@ async def city_detail( return await run_in_threadpool(_refresh_city_market_cache, city, True) cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "market", city) if cached_entry: - if not _city_cache_is_fresh(cached_entry, CITY_MARKET_CACHE_TTL_SEC): + if not _market_analysis_cache_is_fresh(cached_entry): _schedule_cache_refresh(background_tasks, kind="market", city=city) return cached_entry.get("payload") or {} return await run_in_threadpool(_refresh_city_market_cache, city, False) @@ -1526,14 +1634,45 @@ async def city_market_scan( city = _normalize_city_or_404(name) if force_refresh: data = await run_in_threadpool(_refresh_city_market_cache, city, True) + cached_scan = _get_cached_market_scan_payload( + data, + market_slug=market_slug, + target_date=target_date, + ) + if cached_scan is not None: + return cached_scan else: cached_entry = await run_in_threadpool(_CACHE_DB.get_city_cache, "market", city) if cached_entry: - if not _city_cache_is_fresh(cached_entry, CITY_MARKET_CACHE_TTL_SEC): - _schedule_cache_refresh(background_tasks, kind="market", city=city) data = cached_entry.get("payload") or {} + cached_scan = _get_cached_market_scan_payload( + data, + market_slug=market_slug, + target_date=target_date, + ) + if cached_scan is not None: + if not _market_analysis_cache_is_fresh(cached_entry): + _schedule_cache_refresh(background_tasks, kind="market", city=city) + return cached_scan + if _market_analysis_cache_is_fresh(cached_entry): + return await run_in_threadpool( + _refresh_market_scan_payload_from_cached_analysis, + city, + data, + market_slug=market_slug, + target_date=target_date, + ) + else: + _schedule_cache_refresh(background_tasks, kind="market", city=city) else: data = await run_in_threadpool(_refresh_city_market_cache, city, False) + cached_scan = _get_cached_market_scan_payload( + data, + market_slug=market_slug, + target_date=target_date, + ) + if cached_scan is not None: + return cached_scan return await run_in_threadpool( _build_city_market_scan_payload, data,