gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Optimize probability hub market refresh

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-22 04:54:19 +08:00
parent ce394bc089
commit 40564b545d
4 changed files with 326 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/components/probability-hub/ProbabilityHubPage.tsx
+151 -49
View File
@@ -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<string | null>(null);
const [filterMode, setFilterMode] = useState<FilterMode>("all");
const [sortMode, setSortMode] = useState<SortMode>("risk");
const [visibleCityNames, setVisibleCityNames] = useState<Set<string>>(new Set());
const cardNodesRef = useRef(new Map<string, HTMLElement>());
const initialMarketRequestsRef = useRef(new Set<string>());
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<string, CityDetail>,
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<string, CityDetail> = {};
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<string, CityDetail> = {};
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 (
<div className={clsx(dashboardStyles.root, styles.pageRoot)}>
<HeaderBar
@@ -483,7 +575,12 @@ function ProbabilityHubScreen() {
const detail = detailsByName[city.name];
if (!detail) {
return (
<section key={city.name} className={styles.card}>
<section
key={city.name}
ref={(node) => setCardNode(city.name, node)}
data-city-name={city.name}
className={styles.card}
>
<div className={styles.cardHead}>
<div className={styles.cardTitleBlock}>
<div className={styles.cardTitle}>{city.display_name}</div>
@@ -502,7 +599,12 @@ function ProbabilityHubScreen() {
}
return (
<section key={city.name} className={styles.card}>
<section
key={city.name}
ref={(node) => setCardNode(city.name, node)}
data-city-name={city.name}
className={styles.card}
>
<div className={styles.cardHead}>
<div className={styles.cardTitleBlock}>
<div className={styles.cardTitle}>{detail.display_name}</div>
frontend/lib/dashboard-client.ts
+32 -3
View File
@@ -13,6 +13,14 @@ const CACHE_TTL_MS = 5 * 60 * 1000;
const pendingCityDetailRequests = new Map<string, Promise<CityDetail>>();
const pendingHistoryRequests = new Map<string, Promise<HistoryPayload>>();
const pendingCitySummaryRequests = new Map<string, Promise<CitySummary>>();
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 }) {
src/data_collection/polymarket_readonly.py
+1 -3
View File
@@ -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 = {
web/routes.py
+142 -3
View File
@@ -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,