From c79e0f36cce6d0315c59921bd1eced07c14c2c44 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 15 Jun 2026 01:39:19 +0800 Subject: [PATCH] Improve scan terminal freshness and AMSC overlay consistency --- docker-compose.yml | 8 +- frontend/app/api/scan/terminal/route.ts | 2 + .../__tests__/proxyCachePolicy.test.ts | 10 + .../__tests__/refreshCadencePolicy.test.ts | 5 +- .../__tests__/scanTerminalPatchRows.test.ts | 156 +++++++++++++++ .../scan-terminal/scan-terminal-client.ts | 7 + .../scan-terminal/use-scan-terminal-query.ts | 187 ++++++++++++++++-- frontend/lib/dashboard-types.ts | 11 +- frontend/lib/refresh-policy.ts | 2 +- src/utils/refresh_policy.py | 2 +- tests/test_cache_warmer_service.py | 21 +- tests/test_canonical_engine.py | 12 ++ tests/test_deployment_runtime_config.py | 8 +- tests/test_latest_observation_overlay.py | 48 +++++ tests/test_observation_collector.py | 19 ++ tests/test_realtime_patch_schema.py | 12 ++ tests/test_refresh_policy.py | 2 +- tests/test_scan_terminal_modules.py | 141 +++++++++++++ tests/test_web_observability.py | 13 +- web/cache_warmer_service.py | 20 +- web/observation_collector_service.py | 2 +- web/realtime_patch_schema.py | 17 ++ web/routers/scan.py | 4 + web/scan_terminal_cache.py | 8 + web/scan_terminal_payloads.py | 129 ++++++++++++ web/scan_terminal_service.py | 72 ++++++- web/services/canonical_engine.py | 13 +- web/services/canonical_temperature.py | 15 ++ web/services/latest_observation_overlay.py | 9 +- web/services/scan_api.py | 17 ++ 30 files changed, 924 insertions(+), 48 deletions(-) create mode 100644 frontend/components/dashboard/scan-terminal/__tests__/scanTerminalPatchRows.test.ts create mode 100644 tests/test_latest_observation_overlay.py diff --git a/docker-compose.yml b/docker-compose.yml index 483fc7c5..a32f50f8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -217,11 +217,11 @@ services: POLYWEATHER_REDIS_URL: ${POLYWEATHER_REDIS_URL:-redis://polyweather_redis:6379/0} POLYWEATHER_SCAN_TERMINAL_PREWARM_ENABLED: 'false' POLYWEATHER_SERVICE_ROLE: warmer - POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-8} - POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-60} + POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-16} + POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-30} POLYWEATHER_WARMER_ENABLED: ${POLYWEATHER_WARMER_ENABLED:-true} - POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-300} - POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-60} + POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-120} + POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-30} healthcheck: interval: 60s retries: 3 diff --git a/frontend/app/api/scan/terminal/route.ts b/frontend/app/api/scan/terminal/route.ts index 657eb181..a4bcfc9a 100644 --- a/frontend/app/api/scan/terminal/route.ts +++ b/frontend/app/api/scan/terminal/route.ts @@ -43,6 +43,8 @@ export async function GET(req: NextRequest) { "time_range", "limit", "force_refresh", + "diff", + "since_snapshot_id", "timezone_offset_seconds", ]) { const value = req.nextUrl.searchParams.get(key); diff --git a/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts index b72d2885..98ea1081 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/proxyCachePolicy.test.ts @@ -82,6 +82,11 @@ export function runTests() { /fetchCache:\s*"no-store"/, "scan terminal proxy must not put failed or initializing payloads into the Next data cache", ); + assert.match( + scanTerminalProxySource, + /"since_snapshot_id"/, + "scan terminal proxy should forward incremental scan diff snapshot ids to the backend", + ); const scanTerminalClientSource = fs.readFileSync( path.join( @@ -108,6 +113,11 @@ export function runTests() { /params\.set\("_v",\s*SCAN_TERMINAL_PAYLOAD_VERSION\)/, "scan terminal client should vary the CDN cache key without changing backend scan filters", ); + assert.match( + scanTerminalClientSource, + /params\.set\("diff",\s*"true"\)/, + "scan terminal client should request incremental scan diffs when it has a previous snapshot", + ); const overviewProxySource = fs.readFileSync( path.join( diff --git a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts index b8e6529a..94f79d5f 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts @@ -36,7 +36,7 @@ async function flushMicrotasks() { export async function runTests() { assert(DASHBOARD_REFRESH_POLICY_MS.observation === 60_000, "observation layer should refresh every 60 seconds"); - assert(DASHBOARD_REFRESH_POLICY_MS.scanRows === 5 * 60_000, "region/city rows should refresh every 5 minutes"); + assert(DASHBOARD_REFRESH_POLICY_MS.scanRows === 2 * 60_000, "region/city rows should refresh every 2 minutes"); assert(DASHBOARD_REFRESH_POLICY_MS.marketOverview === 10 * 60_000, "market overview should refresh every 10 minutes"); assert(DASHBOARD_REFRESH_POLICY_MS.model === 30 * 60_000, "DEB and multi-model data should refresh every 30 minutes"); assert(DASHBOARD_REFRESH_POLICY_SEC.metar === 5 * 60, "METAR polling should be 5 minutes"); @@ -69,8 +69,9 @@ export async function runTests() { querySource.includes("handleForegroundScanRefresh") && querySource.includes('document.visibilityState !== "visible"') && querySource.includes("SCAN_CACHE_TTL_MS") && + querySource.includes("FOREGROUND_SCAN_REFRESH_AFTER_SUCCESS_MS = 60_000") && querySource.includes("fetchScanTerminal({ forceRefresh: false, showLoading: false })"), - "scan list should silently revalidate stale rows when a long-hidden browser tab returns to the foreground", + "scan list should silently revalidate rows when a browser tab returns to the foreground after 60 seconds", ); assert( chartLogicSource.includes("DASHBOARD_REFRESH_POLICY_MS.metar") && diff --git a/frontend/components/dashboard/scan-terminal/__tests__/scanTerminalPatchRows.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/scanTerminalPatchRows.test.ts new file mode 100644 index 00000000..e9e343ee --- /dev/null +++ b/frontend/components/dashboard/scan-terminal/__tests__/scanTerminalPatchRows.test.ts @@ -0,0 +1,156 @@ +import { + __applyPatchToScanRowForTest, + __mergeScanTerminalIncrementalResponseForTest, +} from "@/components/dashboard/scan-terminal/use-scan-terminal-query"; +import type { ScanTerminalResponse } from "@/lib/dashboard-types"; +import type { CityPatch } from "@/hooks/use-sse-patches"; + +function assert(condition: unknown, message: string) { + if (!condition) throw new Error(message); +} + +export function runTests() { + const patch: CityPatch = { + type: "city_observation_patch.v1", + city: "shenzhen", + revision: 42, + changes: { + temp: 30.2, + max_so_far: 30.4, + signed_gap: 1.4, + gap_to_target: -1.4, + touch_distance: 0, + current_reference: 30.4, + edge: 0.06, + edge_percent: 6, + obs_time: "2026-06-14T06:30:00Z", + }, + }; + + const next = __applyPatchToScanRowForTest( + { + city: "shenzhen", + current_temp: 29.8, + current_max_so_far: 30.0, + signed_gap: 1.0, + gap_to_target: -1.0, + touch_distance: 0.2, + current_reference: 30.0, + edge: 0.04, + edge_percent: 4, + local_time: "old", + target_threshold: 29, + }, + patch, + ) as any; + + assert(next.current_temp === 30.2, "SSE patch should update scan row current temperature"); + assert(next.current_max_so_far === 30.4, "SSE patch should use backend max_so_far instead of only Math.max(row, temp)"); + assert(next.signed_gap === 1.4, "SSE patch should update signed_gap when backend sends it"); + assert(next.gap_to_target === -1.4, "SSE patch should update gap_to_target when backend sends it"); + assert(next.touch_distance === 0, "SSE patch should update touch_distance when backend sends it"); + assert(next.current_reference === 30.4, "SSE patch should update current_reference when backend sends it"); + assert(next.edge === 0.06 && next.edge_percent === 6, "SSE patch should update edge fields when backend sends them"); + assert(next.local_time === "2026-06-14T06:30:00Z", "SSE patch should update observation time"); + assert(next.sse_revision === 42, "SSE patch should record the applied revision"); + + const derived = __applyPatchToScanRowForTest( + { + city: "shenzhen", + current_temp: 28, + current_max_so_far: 28, + target_threshold: 30, + target_label: "30+", + }, + { + type: "city_observation_patch.v1", + city: "shenzhen", + revision: 43, + changes: { + temp: 30.5, + }, + }, + ) as any; + + assert(derived.current_max_so_far === 30.5, "SSE patch should still derive max_so_far from temp when backend omits it"); + assert(derived.signed_gap === 0.5, "SSE patch should derive signed_gap from updated max and row threshold"); + assert(derived.gap_to_target === -0.5, "SSE patch should derive gap_to_target from updated max and row threshold"); + assert(derived.touch_distance === 0, "SSE patch should derive zero touch_distance once threshold is reached"); + + const previous: ScanTerminalResponse = { + generated_at: "2026-06-14T00:00:00Z", + snapshot_id: "scan-old", + status: "ready", + stale: false, + filters: { + scan_mode: "tradable", + min_price: 0.05, + max_price: 0.95, + min_edge_pct: 2, + min_liquidity: 500, + high_liquidity_only: false, + market_type: "maxtemp", + time_range: "today", + limit: 180, + }, + summary: { + recommended_count: 2, + visible_count: 2, + candidate_total: 2, + tradable_market_count: 2, + total_volume: 1000, + }, + top_signal: null, + rows: [ + { id: "row-1", rank: 1, city: "shenzhen", edge_percent: 4 }, + { id: "row-removed", rank: 2, city: "paris", edge_percent: 3 }, + ], + }; + + const notModified = __mergeScanTerminalIncrementalResponseForTest(previous, { + generated_at: "2026-06-14T00:01:00Z", + snapshot_id: "scan-old", + status: "not_modified", + stale: false, + filters: previous.filters, + summary: previous.summary, + top_signal: previous.top_signal, + rows: [], + diff: { + mode: "not_modified", + base_snapshot_id: "scan-old", + snapshot_id: "scan-old", + rows_changed: [], + removed_row_ids: [], + }, + }); + + assert(notModified.rows.length === 2, "not_modified scan diff should keep the previous rows"); + assert(notModified.status === "ready", "not_modified scan diff should normalize to a renderable ready payload"); + + const delta = __mergeScanTerminalIncrementalResponseForTest(previous, { + generated_at: "2026-06-14T00:02:00Z", + snapshot_id: "scan-new", + status: "ready", + stale: false, + filters: previous.filters, + summary: { ...previous.summary, recommended_count: 2 }, + top_signal: { id: "row-added", city: "tokyo" } as any, + rows: [], + diff: { + mode: "row_delta", + base_snapshot_id: "scan-old", + snapshot_id: "scan-new", + rows_changed: [ + { id: "row-1", rank: 1, city: "shenzhen", edge_percent: 6 }, + { id: "row-added", rank: 2, city: "tokyo", edge_percent: 5 }, + ], + removed_row_ids: ["row-removed"], + }, + }); + + assert(delta.snapshot_id === "scan-new", "row_delta scan diff should advance snapshot_id"); + assert(delta.rows.length === 2, "row_delta scan diff should merge changed, added, and removed rows"); + assert(delta.rows[0].id === "row-1" && delta.rows[0].edge_percent === 6, "row_delta scan diff should replace changed rows"); + assert(delta.rows[1].id === "row-added", "row_delta scan diff should append new rows in rank order"); +} diff --git a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts index c8846862..d7a30eac 100644 --- a/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts +++ b/frontend/components/dashboard/scan-terminal/scan-terminal-client.ts @@ -27,6 +27,7 @@ const SCAN_TERMINAL_PAYLOAD_VERSION = "runway-slim-v1"; type TerminalQueryOptions = { forceRefresh?: boolean; signal?: AbortSignal; + sinceSnapshotId?: string | null; timezoneOffsetSeconds?: number | null; tradingRegion?: string; }; @@ -123,6 +124,7 @@ async function readJsonOrThrow(path: string, init?: RequestInit): Promise async function getTerminal({ forceRefresh = false, signal, + sinceSnapshotId, timezoneOffsetSeconds, tradingRegion, }: TerminalQueryOptions = {}) { @@ -143,6 +145,11 @@ async function getTerminal({ if (Number.isFinite(timezoneOffsetSeconds)) { params.set("timezone_offset_seconds", String(Math.trunc(Number(timezoneOffsetSeconds)))); } + const trimmedSnapshotId = String(sinceSnapshotId || "").trim(); + if (!forceRefresh && trimmedSnapshotId) { + params.set("diff", "true"); + params.set("since_snapshot_id", trimmedSnapshotId); + } params.set("_v", SCAN_TERMINAL_PAYLOAD_VERSION); 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 5563019b..c9582720 100644 --- a/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts +++ b/frontend/components/dashboard/scan-terminal/use-scan-terminal-query.ts @@ -13,11 +13,24 @@ import { useSsePatchVersion, type CityPatch, } from "@/hooks/use-sse-patches"; -import type { ScanTerminalResponse } from "@/lib/dashboard-types"; +import type { + ScanOpportunityRow, + ScanTerminalResponse, +} from "@/lib/dashboard-types"; const SCAN_CACHE_PREFIX = "polyweather_scan_v2"; const SCAN_CACHE_TTL_MS = DASHBOARD_REFRESH_POLICY_MS.scanRows; +const FOREGROUND_SCAN_REFRESH_AFTER_SUCCESS_MS = 60_000; const MAX_STALE_SCAN_CACHE_MS = 6 * 60 * 60 * 1000; +const DERIVED_SCAN_PATCH_NUMBER_FIELDS = [ + "signed_gap", + "gap_to_target", + "touch_distance", + "current_reference", + "edge", + "edge_percent", + "deb_prediction", +] as const; function scanCacheKey(tradingRegion: string): string { return `${SCAN_CACHE_PREFIX}:${tradingRegion || "all"}`; @@ -48,23 +61,82 @@ function normalizeCityKey(city: string | null | undefined) { return String(city || "").trim().toLowerCase(); } -function applyPatchToScanRow(row: any, patch: CityPatch) { - const temp = typeof patch.changes.temp === "number" && Number.isFinite(patch.changes.temp) - ? patch.changes.temp - : null; - if (temp === null) return row; +function finiteNumber(value: unknown): number | null { + const number = Number(value); + return Number.isFinite(number) ? number : null; +} + +function firstFiniteNumber(...values: unknown[]) { + for (const value of values) { + const number = finiteNumber(value); + if (number !== null) return number; + } + return null; +} + +function isBelowTargetRow(row: any) { + const text = String( + row?.market_direction || + row?.temperature_direction || + row?.market_question || + row?.target_label || + row?.action || + "", + ).toLowerCase(); + return /\b(below|under|lte|less)\b|<=|≤|以下/.test(text); +} + +function deriveTargetGapFields(row: any, maxSoFar: number | null) { + const target = firstFiniteNumber( + row?.target_threshold, + row?.target_value, + row?.target_lower, + row?.target_upper, + ); + if (target === null || maxSoFar === null) return null; + + const belowTargetRow = isBelowTargetRow(row); + const signedGap = belowTargetRow ? target - maxSoFar : maxSoFar - target; + const touchDistance = belowTargetRow + ? Math.max(0, maxSoFar - target) + : Math.max(0, target - maxSoFar); return { + signed_gap: Number(signedGap.toFixed(2)), + gap_to_target: Number((-signedGap).toFixed(2)), + touch_distance: Number(touchDistance.toFixed(2)), + }; +} + +function applyPatchToScanRow(row: any, patch: CityPatch) { + const temp = finiteNumber(patch.changes.temp); + const patchMaxSoFar = finiteNumber(patch.changes.max_so_far); + if (temp === null) return row; + const currentMaxSoFar = firstFiniteNumber(row.current_max_so_far, temp) ?? temp; + const maxSoFar = patchMaxSoFar ?? Math.max(temp, currentMaxSoFar); + const derivedGapFields = deriveTargetGapFields(row, maxSoFar); + const nextRow: any = { ...row, current_temp: temp, - current_max_so_far: Math.max( - temp, - typeof row.current_max_so_far === "number" && Number.isFinite(row.current_max_so_far) - ? row.current_max_so_far - : temp, - ), + current_max_so_far: maxSoFar, local_time: typeof patch.changes.obs_time === "string" ? patch.changes.obs_time : row.local_time, sse_revision: patch.revision, }; + for (const key of DERIVED_SCAN_PATCH_NUMBER_FIELDS) { + const value = finiteNumber(patch.changes[key]); + if (value !== null) { + nextRow[key] = value; + } + } + if (derivedGapFields) { + for (const key of ["signed_gap", "gap_to_target", "touch_distance"] as const) { + if (finiteNumber(patch.changes[key]) === null) { + nextRow[key] = derivedGapFields[key]; + } + } + } + return { + ...nextRow, + }; } function applyTerminalPatches( @@ -82,6 +154,76 @@ function applyTerminalPatches( return changed ? { ...data, rows } : data; } +function scanRowId(row: ScanOpportunityRow | null | undefined) { + const id = row?.id; + return typeof id === "string" && id ? id : null; +} + +function sortScanRowsByRank(rows: ScanOpportunityRow[]) { + return [...rows].sort((a, b) => { + const rankA = finiteNumber(a.rank); + const rankB = finiteNumber(b.rank); + if (rankA !== null && rankB !== null && rankA !== rankB) { + return rankA - rankB; + } + if (rankA !== null && rankB === null) return -1; + if (rankA === null && rankB !== null) return 1; + return 0; + }); +} + +function mergeScanTerminalIncrementalResponse( + previous: ScanTerminalResponse | null | undefined, + next: ScanTerminalResponse, +): ScanTerminalResponse { + const diff = next.diff; + if (!diff || diff.mode === "full") return next; + if (!previous?.snapshot_id) return next; + + const baseSnapshotId = diff.base_snapshot_id || next.snapshot_id; + if (previous.snapshot_id !== baseSnapshotId) return next; + + if (diff.mode === "not_modified") { + return { + ...previous, + ...next, + status: "ready", + stale: false, + rows: previous.rows, + }; + } + + if (diff.mode !== "row_delta") return next; + + const removedIds = new Set((diff.removed_row_ids || []).map((id) => String(id))); + const changedRows = new Map(); + for (const row of diff.rows_changed || []) { + const id = scanRowId(row); + if (id) changedRows.set(id, row); + } + + const seenIds = new Set(); + const mergedRows: ScanOpportunityRow[] = []; + for (const row of previous.rows || []) { + const id = scanRowId(row); + if (!id || removedIds.has(id)) continue; + const replacement = changedRows.get(id); + mergedRows.push(replacement || row); + seenIds.add(id); + } + for (const [id, row] of changedRows.entries()) { + if (!seenIds.has(id) && !removedIds.has(id)) { + mergedRows.push(row); + } + } + + return { + ...previous, + ...next, + rows: sortScanRowsByRank(mergedRows), + }; +} + export function useScanTerminalQuery({ isPro, proAccessLoading, @@ -114,6 +256,11 @@ export function useScanTerminalQuery({ } return null; }); + const latestScanDataRef = useRef(null); + + useEffect(() => { + latestScanDataRef.current = terminalData || cachedRows; + }, [terminalData, cachedRows]); useEffect(() => { if (typeof window === "undefined") return; @@ -136,13 +283,17 @@ export function useScanTerminalQuery({ lastForcedScanRefreshAtRef.current = Date.now(); } await run({ - request: (signal) => - scanTerminalClient.getTerminal({ + request: async (signal) => { + const previous = latestScanDataRef.current; + const next = await scanTerminalClient.getTerminal({ forceRefresh, signal, + sinceSnapshotId: !forceRefresh ? previous?.snapshot_id : null, timezoneOffsetSeconds, tradingRegion, - }), + }); + return mergeScanTerminalIncrementalResponse(previous, next); + }, showLoading, onSuccess: (data) => { lastScanSuccessAtRef.current = Date.now(); @@ -191,7 +342,7 @@ export function useScanTerminalQuery({ if (now - lastForegroundScanRefreshAtRef.current < 30_000) return; if ( lastScanSuccessAtRef.current > 0 && - now - lastScanSuccessAtRef.current < SCAN_CACHE_TTL_MS + now - lastScanSuccessAtRef.current < FOREGROUND_SCAN_REFRESH_AFTER_SUCCESS_MS ) { return; } @@ -268,3 +419,7 @@ export function useScanTerminalQuery({ terminalData: effectiveData, }; } + +export const __applyPatchToScanRowForTest = applyPatchToScanRow; +export const __mergeScanTerminalIncrementalResponseForTest = + mergeScanTerminalIncrementalResponse; diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index 4336a3ec..a28e2799 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -507,6 +507,14 @@ export interface ScanTerminalFilters { limit: number; } +export interface ScanTerminalDiff { + mode: "full" | "not_modified" | "row_delta" | string; + base_snapshot_id?: string | null; + snapshot_id?: string | null; + rows_changed?: ScanOpportunityRow[]; + removed_row_ids?: string[]; +} + export interface ScanOpportunityRow { id: string; rank?: number | null; @@ -673,7 +681,7 @@ export interface PrimarySignal extends ScanOpportunityRow {} export interface ScanTerminalResponse { generated_at: string; snapshot_id?: string | null; - status?: "ready" | "stale" | "failed" | string; + status?: "ready" | "stale" | "failed" | "not_modified" | string; stale?: boolean; stale_reason?: string | null; last_success_at?: string | null; @@ -694,6 +702,7 @@ export interface ScanTerminalResponse { }; top_signal?: PrimarySignal | null; rows: ScanOpportunityRow[]; + diff?: ScanTerminalDiff; } export interface IntradayMeteorologySignal { diff --git a/frontend/lib/refresh-policy.ts b/frontend/lib/refresh-policy.ts index f8f05f34..aaf618b5 100644 --- a/frontend/lib/refresh-policy.ts +++ b/frontend/lib/refresh-policy.ts @@ -1,7 +1,7 @@ export const DASHBOARD_REFRESH_POLICY_SEC = { observation: 60, metar: 5 * 60, - scanRows: 5 * 60, + scanRows: 2 * 60, marketOverview: 10 * 60, model: 30 * 60, } as const; diff --git a/src/utils/refresh_policy.py b/src/utils/refresh_policy.py index 55a37047..00105bcd 100644 --- a/src/utils/refresh_policy.py +++ b/src/utils/refresh_policy.py @@ -4,6 +4,6 @@ from __future__ import annotations OBSERVATION_REFRESH_SEC = 60 METAR_POLL_TTL_SEC = 5 * 60 -SCAN_ROWS_REFRESH_SEC = 5 * 60 +SCAN_ROWS_REFRESH_SEC = 2 * 60 MARKET_OVERVIEW_TTL_SEC = 10 * 60 MODEL_CACHE_TTL_SEC = 30 * 60 diff --git a/tests/test_cache_warmer_service.py b/tests/test_cache_warmer_service.py index ee03317e..5bf1f6ae 100644 --- a/tests/test_cache_warmer_service.py +++ b/tests/test_cache_warmer_service.py @@ -1,6 +1,11 @@ from datetime import datetime, timezone -from web.cache_warmer_service import CacheWarmer, build_default_cache_warmer, build_priority_city_batch +from web.cache_warmer_service import ( + CacheWarmer, + build_default_cache_warmer, + build_priority_city_batch, + warmer_tick_sec, +) def test_priority_city_batch_prefers_local_active_hours_over_night(): @@ -82,6 +87,20 @@ def test_default_cache_warmer_enqueues_city_refresh_without_direct_panel_refresh ] +def test_default_cache_warmer_uses_faster_city_refresh_defaults(monkeypatch): + monkeypatch.delenv("POLYWEATHER_WARMER_CITY_BATCH_SIZE", raising=False) + monkeypatch.delenv("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", raising=False) + monkeypatch.delenv("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", raising=False) + monkeypatch.delenv("POLYWEATHER_WARMER_TICK_SEC", raising=False) + + warmer = build_default_cache_warmer() + + assert warmer.scan_interval_sec == 120 + assert warmer.city_batch_size == 16 + assert warmer.city_interval_sec == 30 + assert warmer_tick_sec() == 30 + + def test_cache_warmer_skips_work_when_intervals_are_not_due(): now_ts = datetime(2026, 6, 14, 12, 0, tzinfo=timezone.utc).timestamp() scan_calls = [] diff --git a/tests/test_canonical_engine.py b/tests/test_canonical_engine.py index cba4a34d..8efa16f1 100644 --- a/tests/test_canonical_engine.py +++ b/tests/test_canonical_engine.py @@ -98,6 +98,12 @@ def test_canonical_engine_builds_realtime_event_from_canonical(): "freshness_sec": 300, "freshness_status": "fresh", "confidence": 0.9, + "max_so_far": 28.6, + "signed_gap": 1.2, + "gap_to_target": -1.2, + "touch_distance": 0.0, + "edge": 0.07, + "edge_percent": 7.0, } ) @@ -112,3 +118,9 @@ def test_canonical_engine_builds_realtime_event_from_canonical(): assert event["payload"]["unit"] == "celsius" assert event["payload"]["freshness_status"] == "fresh" assert event["payload"]["confidence"] == 0.9 + assert event["payload"]["max_so_far"] == 28.6 + assert event["payload"]["signed_gap"] == 1.2 + assert event["payload"]["gap_to_target"] == -1.2 + assert event["payload"]["touch_distance"] == 0.0 + assert event["payload"]["edge"] == 0.07 + assert event["payload"]["edge_percent"] == 7.0 diff --git a/tests/test_deployment_runtime_config.py b/tests/test_deployment_runtime_config.py index 6284078c..2cb2fb83 100644 --- a/tests/test_deployment_runtime_config.py +++ b/tests/test_deployment_runtime_config.py @@ -127,10 +127,10 @@ def test_docker_compose_isolates_collector_from_web_and_bot_services(): assert "POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_AMSC_SEC:-60}" in collector_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS: ${POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS:-2}" in collector_block assert "POLYWEATHER_WARMER_ENABLED: ${POLYWEATHER_WARMER_ENABLED:-true}" in warmer_block - assert "POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-60}" in warmer_block - assert "POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-300}" in warmer_block - assert "POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-60}" in warmer_block - assert "POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-8}" in warmer_block + assert "POLYWEATHER_WARMER_TICK_SEC: ${POLYWEATHER_WARMER_TICK_SEC:-30}" in warmer_block + assert "POLYWEATHER_WARMER_SCAN_INTERVAL_SEC: ${POLYWEATHER_WARMER_SCAN_INTERVAL_SEC:-120}" in warmer_block + assert "POLYWEATHER_WARMER_CITY_INTERVAL_SEC: ${POLYWEATHER_WARMER_CITY_INTERVAL_SEC:-30}" in warmer_block + assert "POLYWEATHER_WARMER_CITY_BATCH_SIZE: ${POLYWEATHER_WARMER_CITY_BATCH_SIZE:-16}" in warmer_block assert "cpus: ${POLYWEATHER_WARMER_CPUS:-0.75}" in warmer_block assert "TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC: ${POLYWEATHER_BOT_AIRPORT_PUSH_INTERVAL_SEC:-60}" in bot_block assert "POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC: ${POLYWEATHER_OBSERVATION_COLLECTOR_MADIS_SEC:-300}" in collector_block diff --git a/tests/test_latest_observation_overlay.py b/tests/test_latest_observation_overlay.py new file mode 100644 index 00000000..44176447 --- /dev/null +++ b/tests/test_latest_observation_overlay.py @@ -0,0 +1,48 @@ +from web.services.latest_observation_overlay import overlay_latest_amsc_observation + + +def test_overlay_replaces_amos_when_old_local_time_string_looks_later_than_new_utc(): + class FakeDB: + def get_latest_raw_observation(self, source, city): + assert (source, city) == ("amsc_awos", "qingdao") + return { + "observed_at": "2026-06-14T17:23:00+00:00", + "fetched_at": "2026-06-14T17:23:30+00:00", + "station_code": "ZSQD", + "station_name": "Qingdao Jiaodong", + "payload": { + "source": "amsc_awos", + "source_label": "AMSC AWOS Qingdao Jiaodong (ZSQD)", + "icao": "ZSQD", + "temp_c": 18.2, + "observation_time": "2026-06-14T17:23:00+00:00", + "observation_time_local": "2026-06-15 01:23:00", + "runway_obs": { + "point_temperatures": [ + { + "runway": "17L/35R", + "tdz_temp": 18.0, + "mid_temp": 18.2, + "end_temp": 18.1, + } + ], + }, + }, + } + + stale_payload = { + "name": "qingdao", + "temp_symbol": "C", + "amos": { + "source": "amsc_awos", + "temp_c": 21.0, + "observation_time": "2026-06-14T10:46:00+00:00", + "observation_time_local": "2026-06-14 18:46:00", + }, + } + + result = overlay_latest_amsc_observation(FakeDB(), "qingdao", stale_payload) + + assert result["amos"]["observation_time"] == "2026-06-14T17:23:00+00:00" + assert result["amos"]["observation_time_local"] == "2026-06-15 01:23:00" + assert result["amos"]["runway_obs"]["point_temperatures"][0]["runway"] == "17L/35R" diff --git a/tests/test_observation_collector.py b/tests/test_observation_collector.py index 730a50f6..1e2e969a 100644 --- a/tests/test_observation_collector.py +++ b/tests/test_observation_collector.py @@ -132,6 +132,25 @@ def test_observation_collector_run_due_once_collects_without_panel_cache_refresh assert calls == [("qingdao", False), ("qingdao", False)] +def test_observation_collector_default_cache_refresh_workers_is_two(monkeypatch): + from web.observation_collector_service import ObservationCollector + + monkeypatch.delenv( + "POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", + raising=False, + ) + collector = ObservationCollector( + weather=object(), + profiles=[], + cache_refresher=lambda _city: None, + ) + try: + assert collector._cache_refresh_executor is not None + assert collector._cache_refresh_executor._max_workers == 2 + finally: + collector._cache_refresh_executor.shutdown(wait=False) + + def test_raw_observation_store_records_latest_observation(tmp_path): from src.database.db_manager import DBManager diff --git a/tests/test_realtime_patch_schema.py b/tests/test_realtime_patch_schema.py index 9ff87e65..e1be457d 100644 --- a/tests/test_realtime_patch_schema.py +++ b/tests/test_realtime_patch_schema.py @@ -66,6 +66,12 @@ def test_v1_patch_payload_is_accepted_and_normalized(): "obs_time": "2026-05-26T07:01:00Z", "payload": { "temp": 29.4, + "max_so_far": 30.1, + "signed_gap": 0.6, + "gap_to_target": -0.6, + "touch_distance": 0, + "edge": 0.04, + "edge_percent": 4.0, "station_code": "46692", "runway_points": [{"runway": "05/23", "temp": 30.2}], }, @@ -75,6 +81,12 @@ def test_v1_patch_payload_is_accepted_and_normalized(): assert event["city"] == "taipei" assert event["source"] == "cwa" assert event["payload"]["temp"] == 29.4 + assert event["payload"]["max_so_far"] == 30.1 + assert event["payload"]["signed_gap"] == 0.6 + assert event["payload"]["gap_to_target"] == -0.6 + assert event["payload"]["touch_distance"] == 0 + assert event["payload"]["edge"] == 0.04 + assert event["payload"]["edge_percent"] == 4.0 assert event["payload"]["runway_points"][0]["temp"] == 30.2 diff --git a/tests/test_refresh_policy.py b/tests/test_refresh_policy.py index dc50df5e..169ce516 100644 --- a/tests/test_refresh_policy.py +++ b/tests/test_refresh_policy.py @@ -10,7 +10,7 @@ from src.utils.refresh_policy import ( def test_refresh_policy_cadences_are_layered(): assert OBSERVATION_REFRESH_SEC == 60 assert METAR_POLL_TTL_SEC == 300 - assert SCAN_ROWS_REFRESH_SEC == 300 + assert SCAN_ROWS_REFRESH_SEC == 120 assert MARKET_OVERVIEW_TTL_SEC == 600 assert MODEL_CACHE_TTL_SEC == 1800 diff --git a/tests/test_scan_terminal_modules.py b/tests/test_scan_terminal_modules.py index d08d185e..0be8dd7e 100644 --- a/tests/test_scan_terminal_modules.py +++ b/tests/test_scan_terminal_modules.py @@ -3,6 +3,7 @@ from web import scan_terminal_cache from web import scan_terminal_service from web.scan_terminal_metar_gate import _apply_metar_gate_to_row from web.scan_terminal_payloads import ( + build_scan_terminal_incremental_payload, build_failed_scan_terminal_payload, build_scan_terminal_snapshot_id, build_stale_scan_terminal_payload, @@ -76,6 +77,146 @@ def test_scan_terminal_failure_state_preserves_redis_success_payload(monkeypatch assert entry["last_error"] == "timeout" +def test_scan_terminal_cache_keeps_previous_success_snapshot_for_diffs(monkeypatch): + fake_redis = _FakeRedis() + monkeypatch.setenv("POLYWEATHER_SCAN_TERMINAL_REDIS_CACHE_ENABLED", "true") + monkeypatch.setattr(scan_terminal_cache, "_get_redis_client", lambda: fake_redis) + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + + filters = {"scan_mode": "tradable", "limit": 9} + scan_terminal_cache.set_cached_scan_terminal_payload( + filters, + { + "snapshot_id": "scan-old", + "generated_at": "2026-06-01T00:00:00Z", + "rows": [{"id": "row-1", "edge_percent": 3}], + }, + ) + scan_terminal_cache.set_cached_scan_terminal_payload( + filters, + { + "snapshot_id": "scan-new", + "generated_at": "2026-06-01T00:01:00Z", + "rows": [{"id": "row-1", "edge_percent": 4}], + }, + ) + + scan_terminal_cache._SCAN_TERMINAL_CACHE.clear() + entry = scan_terminal_cache.get_scan_terminal_cache_entry(filters) + + assert entry["success_payload"]["snapshot_id"] == "scan-new" + assert entry["previous_success_payload"]["snapshot_id"] == "scan-old" + + +def test_build_scan_terminal_incremental_payload_returns_not_modified(): + filters = {"scan_mode": "tradable", "limit": 2} + current = { + "generated_at": "2026-06-01T00:01:00Z", + "snapshot_id": "scan-current", + "status": "ready", + "stale": False, + "filters": filters, + "summary": {"candidate_total": 2}, + "top_signal": {"id": "row-1"}, + "rows": [{"id": "row-1"}, {"id": "row-2"}], + } + + payload = build_scan_terminal_incremental_payload( + filters=filters, + current_payload=current, + since_snapshot_id="scan-current", + base_payload=current, + ) + + assert payload["status"] == "not_modified" + assert payload["rows"] == [] + assert payload["summary"] == current["summary"] + assert payload["top_signal"] == current["top_signal"] + assert payload["diff"] == { + "mode": "not_modified", + "base_snapshot_id": "scan-current", + "snapshot_id": "scan-current", + "rows_changed": [], + "removed_row_ids": [], + } + + +def test_build_scan_terminal_incremental_payload_returns_changed_row_delta(): + filters = {"scan_mode": "tradable", "limit": 3} + base = { + "generated_at": "2026-06-01T00:00:00Z", + "snapshot_id": "scan-old", + "status": "ready", + "stale": False, + "filters": filters, + "summary": {"candidate_total": 2}, + "top_signal": {"id": "row-1"}, + "rows": [ + {"id": "row-1", "rank": 1, "edge_percent": 3}, + {"id": "row-removed", "rank": 2, "edge_percent": 2}, + ], + } + current = { + "generated_at": "2026-06-01T00:01:00Z", + "snapshot_id": "scan-new", + "status": "ready", + "stale": False, + "filters": filters, + "summary": {"candidate_total": 2}, + "top_signal": {"id": "row-added"}, + "rows": [ + {"id": "row-1", "rank": 1, "edge_percent": 4}, + {"id": "row-added", "rank": 2, "edge_percent": 5}, + ], + } + + payload = build_scan_terminal_incremental_payload( + filters=filters, + current_payload=current, + since_snapshot_id="scan-old", + base_payload=base, + ) + + assert payload["status"] == "ready" + assert payload["rows"] == [] + assert payload["snapshot_id"] == "scan-new" + assert payload["diff"]["mode"] == "row_delta" + assert payload["diff"]["base_snapshot_id"] == "scan-old" + assert payload["diff"]["snapshot_id"] == "scan-new" + assert {row["id"] for row in payload["diff"]["rows_changed"]} == { + "row-1", + "row-added", + } + assert payload["diff"]["removed_row_ids"] == ["row-removed"] + + +def test_build_scan_terminal_incremental_payload_falls_back_to_full_without_base(): + filters = {"scan_mode": "tradable", "limit": 2} + current = { + "generated_at": "2026-06-01T00:01:00Z", + "snapshot_id": "scan-new", + "status": "ready", + "stale": False, + "filters": filters, + "summary": {"candidate_total": 1}, + "top_signal": None, + "rows": [{"id": "row-1", "edge_percent": 4}], + } + + payload = build_scan_terminal_incremental_payload( + filters=filters, + current_payload=current, + since_snapshot_id="scan-old", + base_payload=None, + ) + + assert payload["status"] == "ready" + assert payload["rows"] == current["rows"] + assert payload["diff"]["mode"] == "full" + assert payload["diff"]["base_snapshot_id"] == "scan-old" + assert payload["diff"]["snapshot_id"] == "scan-new" + + def test_scan_terminal_prewarm_covers_default_api_limit(): limits = {filters["limit"] for filters in _scan_terminal_prewarm_filters()} diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 3869ddd9..0304f0fb 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -3851,9 +3851,17 @@ def test_scan_terminal_endpoint_forwards_filters(monkeypatch): captured = {} - def _fake_build_scan_terminal_payload(filters, *, force_refresh=False): + def _fake_build_scan_terminal_payload( + filters, + *, + force_refresh=False, + diff=False, + since_snapshot_id=None, + ): captured["filters"] = dict(filters) captured["force_refresh"] = force_refresh + captured["diff"] = diff + captured["since_snapshot_id"] = since_snapshot_id return { "generated_at": "2026-04-23T00:00:00Z", "filters": filters, @@ -3876,12 +3884,15 @@ def test_scan_terminal_endpoint_forwards_filters(monkeypatch): response = client.get( "/api/scan/terminal?scan_mode=trend&min_price=0.1&max_price=0.8&min_edge_pct=3" "&min_liquidity=700&high_liquidity_only=true&market_type=all&time_range=week&limit=12&force_refresh=true" + "&diff=true&since_snapshot_id=scan-old" ) assert response.status_code == 200 payload = response.json() assert payload["summary"]["recommended_count"] == 1 assert captured["force_refresh"] is True + assert captured["diff"] is True + assert captured["since_snapshot_id"] == "scan-old" assert captured["filters"]["scan_mode"] == "trend" assert captured["filters"]["market_type"] == "all" assert captured["filters"]["time_range"] == "week" diff --git a/web/cache_warmer_service.py b/web/cache_warmer_service.py index 8e58981c..51040d7a 100644 --- a/web/cache_warmer_service.py +++ b/web/cache_warmer_service.py @@ -190,17 +190,17 @@ class CacheWarmer: city_provider: Callable[[], Mapping[str, Mapping[str, Any]]], scan_warmer: Callable[..., Any], city_panel_warmer: Callable[..., Any], - scan_interval_sec: int = 300, - city_interval_sec: int = 60, - city_batch_size: int = 8, + scan_interval_sec: int = 120, + city_interval_sec: int = 30, + city_batch_size: int = 16, hot_cities: Optional[Iterable[str]] = None, ) -> None: self.city_provider = city_provider self.scan_warmer = scan_warmer self.city_panel_warmer = city_panel_warmer - self.scan_interval_sec = max(60, int(scan_interval_sec or 300)) - self.city_interval_sec = max(30, int(city_interval_sec or 60)) - self.city_batch_size = max(1, min(32, int(city_batch_size or 8))) + self.scan_interval_sec = max(60, int(scan_interval_sec or 120)) + self.city_interval_sec = max(30, int(city_interval_sec or 30)) + self.city_batch_size = max(1, min(32, int(city_batch_size or 16))) self.hot_cities = tuple(hot_cities or DEFAULT_HOT_CITIES) self._last_scan_ts = 0.0 self._last_city_ts = 0.0 @@ -279,9 +279,9 @@ def build_default_cache_warmer() -> CacheWarmer: city_provider=lambda: CITIES, scan_warmer=build_scan_terminal_payload, city_panel_warmer=_queue_city_panel_refresh, - scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 300), - city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 60), - city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 8), + scan_interval_sec=_env_int("POLYWEATHER_WARMER_SCAN_INTERVAL_SEC", 120), + city_interval_sec=_env_int("POLYWEATHER_WARMER_CITY_INTERVAL_SEC", 30), + city_batch_size=_env_int("POLYWEATHER_WARMER_CITY_BATCH_SIZE", 16), hot_cities=hot_cities or DEFAULT_HOT_CITIES, ) @@ -291,4 +291,4 @@ def warmer_enabled() -> bool: def warmer_tick_sec() -> int: - return max(10, _env_int("POLYWEATHER_WARMER_TICK_SEC", 60)) + return max(10, _env_int("POLYWEATHER_WARMER_TICK_SEC", 30)) diff --git a/web/observation_collector_service.py b/web/observation_collector_service.py index 73e6a431..7902d2c9 100644 --- a/web/observation_collector_service.py +++ b/web/observation_collector_service.py @@ -96,7 +96,7 @@ class ObservationCollector: int( cache_refresh_workers if cache_refresh_workers is not None - else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 1) + else _env_int("POLYWEATHER_OBSERVATION_COLLECTOR_CACHE_REFRESH_WORKERS", 2) ), ), ) diff --git a/web/realtime_patch_schema.py b/web/realtime_patch_schema.py index d0bf5550..cfefcb51 100644 --- a/web/realtime_patch_schema.py +++ b/web/realtime_patch_schema.py @@ -16,6 +16,15 @@ from src.data_collection.city_time import ( SCHEMA_TYPE = "city_observation_patch" SCHEMA_VERSION = 1 EVENT_TYPE = "city_observation_patch.v1" +DERIVED_SCAN_NUMBER_FIELDS = ( + "signed_gap", + "gap_to_target", + "touch_distance", + "current_reference", + "edge", + "edge_percent", + "deb_prediction", +) SOURCE_CADENCE_SECONDS = { "amos": 60, "amsc_awos": 180, @@ -178,6 +187,10 @@ def _payload_from_legacy(changes: Dict[str, Any]) -> Dict[str, Any]: max_so_far = _first_number(changes.get("max_so_far"), changes.get("current_max_so_far")) if max_so_far is not None: payload["max_so_far"] = max_so_far + for key in DERIVED_SCAN_NUMBER_FIELDS: + value = _finite_number(changes.get(key)) + if value is not None: + payload[key] = value station_code = str( changes.get("station_code") @@ -227,6 +240,10 @@ def _payload_from_v1(raw_payload: Any) -> Dict[str, Any]: max_so_far = _finite_number(raw_payload.get("max_so_far")) if max_so_far is not None: payload["max_so_far"] = max_so_far + for key in DERIVED_SCAN_NUMBER_FIELDS: + value = _finite_number(raw_payload.get(key)) + if value is not None: + payload[key] = value for key in ("station_code", "station_label", "series_key", "unit"): value = raw_payload.get(key) diff --git a/web/routers/scan.py b/web/routers/scan.py index 78366c0f..3244d4a8 100644 --- a/web/routers/scan.py +++ b/web/routers/scan.py @@ -30,6 +30,8 @@ async def scan_terminal( time_range: str = "today", limit: int = 25, force_refresh: bool = False, + diff: bool = False, + since_snapshot_id: str | None = None, region: str = "", trading_region: str = "", timezone_offset_seconds: int | None = None, @@ -46,6 +48,8 @@ async def scan_terminal( time_range=time_range, limit=limit, force_refresh=force_refresh, + diff=diff, + since_snapshot_id=since_snapshot_id, region=region or trading_region or None, timezone_offset_seconds=timezone_offset_seconds, ) diff --git a/web/scan_terminal_cache.py b/web/scan_terminal_cache.py index bb0b486c..8da451e4 100644 --- a/web/scan_terminal_cache.py +++ b/web/scan_terminal_cache.py @@ -154,11 +154,19 @@ def set_cached_scan_terminal_payload( cache_key = scan_terminal_cache_key(filters) existing = get_scan_terminal_cache_entry(filters) or {} now = time.time() + existing_success_payload = existing.get("success_payload") + previous_success_payload = existing.get("previous_success_payload") + if isinstance(existing_success_payload, dict): + existing_snapshot_id = existing_success_payload.get("snapshot_id") + next_snapshot_id = payload.get("snapshot_id") + if existing_snapshot_id and existing_snapshot_id != next_snapshot_id: + previous_success_payload = dict(existing_success_payload) entry = { "t": now, "payload": dict(payload), "success_t": now, "success_payload": dict(payload), + "previous_success_payload": previous_success_payload, "last_error": existing.get("last_error"), "last_failed_at": existing.get("last_failed_at"), } diff --git a/web/scan_terminal_payloads.py b/web/scan_terminal_payloads.py index c07f6364..04da1845 100644 --- a/web/scan_terminal_payloads.py +++ b/web/scan_terminal_payloads.py @@ -81,6 +81,135 @@ def build_scan_terminal_snapshot_id( return f"scan-{digest[:10]}" +def _scan_row_id(row: Any) -> Optional[str]: + if not isinstance(row, dict): + return None + row_id = row.get("id") + if row_id is None: + return None + text = str(row_id) + return text if text else None + + +def _scan_row_digest(row: Dict[str, Any]) -> str: + return hashlib.sha256( + json.dumps(row, ensure_ascii=True, sort_keys=True, default=str).encode("utf-8") + ).hexdigest() + + +def _rows_by_id(payload: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: + rows = payload.get("rows") + if not isinstance(rows, list): + return {} + indexed: Dict[str, Dict[str, Any]] = {} + for row in rows: + if not isinstance(row, dict): + continue + row_id = _scan_row_id(row) + if row_id is not None: + indexed[row_id] = row + return indexed + + +def _full_scan_terminal_incremental_payload( + *, + current_payload: Dict[str, Any], + since_snapshot_id: str, +) -> Dict[str, Any]: + payload = dict(current_payload) + payload["diff"] = { + "mode": "full", + "base_snapshot_id": since_snapshot_id, + "snapshot_id": current_payload.get("snapshot_id"), + "rows_changed": [], + "removed_row_ids": [], + } + return payload + + +def build_scan_terminal_incremental_payload( + *, + filters: Dict[str, Any], + current_payload: Dict[str, Any], + since_snapshot_id: Optional[str], + base_payload: Optional[Dict[str, Any]], +) -> Dict[str, Any]: + since_snapshot_id = str(since_snapshot_id or "").strip() + current_snapshot_id = current_payload.get("snapshot_id") + if not since_snapshot_id or not current_snapshot_id: + return dict(current_payload) + if current_payload.get("stale") is True or current_payload.get("status") not in ( + "ready", + "partial", + ): + return _full_scan_terminal_incremental_payload( + current_payload=current_payload, + since_snapshot_id=since_snapshot_id, + ) + if since_snapshot_id == current_snapshot_id: + return { + "generated_at": current_payload.get("generated_at"), + "snapshot_id": current_snapshot_id, + "status": "not_modified", + "stale": False, + "stale_reason": None, + "last_success_at": current_payload.get("last_success_at"), + "last_failed_at": current_payload.get("last_failed_at"), + "filters": filters, + "summary": current_payload.get("summary") or {}, + "top_signal": current_payload.get("top_signal"), + "rows": [], + "diff": { + "mode": "not_modified", + "base_snapshot_id": since_snapshot_id, + "snapshot_id": current_snapshot_id, + "rows_changed": [], + "removed_row_ids": [], + }, + } + + if ( + not isinstance(base_payload, dict) + or base_payload.get("snapshot_id") != since_snapshot_id + ): + return _full_scan_terminal_incremental_payload( + current_payload=current_payload, + since_snapshot_id=since_snapshot_id, + ) + + current_rows = _rows_by_id(current_payload) + base_rows = _rows_by_id(base_payload) + rows_changed: List[Dict[str, Any]] = [] + for row_id, row in current_rows.items(): + base_row = base_rows.get(row_id) + if base_row is None or _scan_row_digest(base_row) != _scan_row_digest(row): + rows_changed.append(row) + + removed_row_ids = [ + row_id for row_id in base_rows.keys() if row_id not in current_rows + ] + return { + "generated_at": current_payload.get("generated_at"), + "snapshot_id": current_snapshot_id, + "status": current_payload.get("status") or "ready", + "stale": False, + "stale_reason": current_payload.get("stale_reason"), + "last_success_at": current_payload.get("last_success_at"), + "last_failed_at": current_payload.get("last_failed_at"), + "filters": filters, + "summary": current_payload.get("summary") or {}, + "top_signal": current_payload.get("top_signal"), + "rows": [], + "diff": { + "mode": "row_delta", + "base_snapshot_id": since_snapshot_id, + "snapshot_id": current_snapshot_id, + "rows_changed": rows_changed, + "removed_row_ids": removed_row_ids, + }, + } + + def build_stale_scan_terminal_payload( *, filters: Dict[str, Any], diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index 7eb3dd32..dcff5f4b 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -35,6 +35,7 @@ from web.scan_terminal_filters import ( ) from web.scan_terminal_payloads import ( build_failed_scan_terminal_payload, + build_scan_terminal_incremental_payload, build_scan_terminal_snapshot_id, build_stale_scan_terminal_payload, compact_ranked_scan_rows_for_payload, @@ -75,6 +76,36 @@ def _success_payload_within_stale_window(cached_entry: Dict[str, Any]) -> bool: return (time.time() - success_ts) <= float(SCAN_TERMINAL_STALE_SUCCESS_MAX_AGE_SEC) +def _build_scan_terminal_incremental_from_cache_entry( + *, + filters: Dict[str, Any], + current_payload: Dict[str, Any], + cached_entry: Optional[Dict[str, Any]], + since_snapshot_id: Optional[str], +) -> Dict[str, Any]: + base_payload: Optional[Dict[str, Any]] = None + since_snapshot_id = str(since_snapshot_id or "").strip() + candidates: List[Any] = [current_payload] + if isinstance(cached_entry, dict): + candidates.extend( + [ + cached_entry.get("success_payload"), + cached_entry.get("previous_success_payload"), + cached_entry.get("payload"), + ] + ) + for candidate in candidates: + if isinstance(candidate, dict) and candidate.get("snapshot_id") == since_snapshot_id: + base_payload = candidate + break + return build_scan_terminal_incremental_payload( + filters=filters, + current_payload=current_payload, + since_snapshot_id=since_snapshot_id, + base_payload=base_payload, + ) + + def _build_stale_payload_for_timeout_if_better_cached( *, filters: Dict[str, Any], @@ -339,6 +370,8 @@ def build_scan_terminal_payload( raw_filters: Optional[Dict[str, Any]] = None, *, force_refresh: bool = False, + diff: bool = False, + since_snapshot_id: Optional[str] = None, timing_recorder: Any = None, ) -> Dict[str, Any]: filters = ( @@ -364,6 +397,14 @@ def build_scan_terminal_payload( ) ) if cached is not None: + if diff and since_snapshot_id: + cached_entry = get_scan_terminal_cache_entry(filters) or {} + return _build_scan_terminal_incremental_from_cache_entry( + filters=filters, + current_payload=cached, + cached_entry=cached_entry, + since_snapshot_id=since_snapshot_id, + ) return cached cached_entry = ( @@ -381,7 +422,7 @@ def build_scan_terminal_payload( and _success_payload_within_stale_window(cached_entry) ): started = _start_scan_terminal_background_refresh(filters) - return build_stale_scan_terminal_payload( + payload = build_stale_scan_terminal_payload( filters=filters, success_payload=success_payload, error_message=( @@ -389,8 +430,16 @@ def build_scan_terminal_payload( ), failed_at=cached_entry.get("last_failed_at"), ) + if diff and since_snapshot_id: + return _build_scan_terminal_incremental_from_cache_entry( + filters=filters, + current_payload=payload, + cached_entry=cached_entry, + since_snapshot_id=since_snapshot_id, + ) + return payload started = _start_scan_terminal_background_refresh(filters) - return build_failed_scan_terminal_payload( + payload = build_failed_scan_terminal_payload( filters=filters, error_message=( "市场扫描快照正在初始化" @@ -399,8 +448,16 @@ def build_scan_terminal_payload( ), failed_at=cached_entry.get("last_failed_at"), ) + if diff and since_snapshot_id: + return _build_scan_terminal_incremental_from_cache_entry( + filters=filters, + current_payload=payload, + cached_entry=cached_entry, + since_snapshot_id=since_snapshot_id, + ) + return payload - return ( + payload = ( timing_recorder.measure( "uncached_build", lambda: _build_scan_terminal_payload_singleflight( @@ -411,6 +468,15 @@ def build_scan_terminal_payload( if timing_recorder is not None else _build_scan_terminal_payload_singleflight(filters, force_refresh=force_refresh) ) + if diff and since_snapshot_id: + cached_entry = get_scan_terminal_cache_entry(filters) or {} + return _build_scan_terminal_incremental_from_cache_entry( + filters=filters, + current_payload=payload, + cached_entry=cached_entry, + since_snapshot_id=since_snapshot_id, + ) + return payload _SCAN_PREWARM_STARTED = False diff --git a/web/services/canonical_engine.py b/web/services/canonical_engine.py index 20813550..4b36b1e8 100644 --- a/web/services/canonical_engine.py +++ b/web/services/canonical_engine.py @@ -185,7 +185,18 @@ def build_realtime_event_from_canonical(canonical: dict[str, Any]) -> Optional[d text = str(canonical.get(key) or "").strip() if text: payload[key] = text - for key in ("freshness_sec", "confidence"): + for key in ( + "freshness_sec", + "confidence", + "max_so_far", + "signed_gap", + "gap_to_target", + "touch_distance", + "current_reference", + "edge", + "edge_percent", + "deb_prediction", + ): value_meta = canonical.get(key) if value_meta is not None and value_meta != "": payload[key] = value_meta diff --git a/web/services/canonical_temperature.py b/web/services/canonical_temperature.py index 4b805197..03dc4473 100644 --- a/web/services/canonical_temperature.py +++ b/web/services/canonical_temperature.py @@ -28,6 +28,14 @@ def _to_int(value: Any) -> Optional[int]: return None +def _first_float(*values: Any) -> Optional[float]: + for value in values: + number = _to_float(value) + if number is not None: + return number + return None + + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -127,6 +135,7 @@ def build_canonical_temperature( canonical = { "city": str(city or payload.get("name") or payload.get("city") or "").strip().lower(), "value": round(value, 2), + "max_so_far": _first_float(current.get("max_so_far"), current.get("max_temp_so_far")), "temp_symbol": str(payload.get("temp_symbol") or "°C"), "source": source, "source_label": source_label, @@ -142,6 +151,10 @@ def build_canonical_temperature( } age_text = f" updated {freshness_sec}s ago" if freshness_sec is not None else "" canonical["explanation"] = f"{source_label or source or 'Source'}{age_text}." + deb = payload.get("deb") if isinstance(payload.get("deb"), dict) else {} + deb_prediction = _to_float(deb.get("prediction")) + if deb_prediction is not None: + canonical["deb_prediction"] = deb_prediction return canonical @@ -190,6 +203,7 @@ def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> O } current = { "temp": value, + "max_so_far": _first_float(canonical.get("max_so_far"), value), "source_code": source, "settlement_source": source, "settlement_source_label": source_label, @@ -201,6 +215,7 @@ def build_city_weather_from_canonical(city: str, canonical: Dict[str, Any]) -> O } airport_primary = { "temp": value, + "max_so_far": _first_float(canonical.get("max_so_far"), value), "source_code": source, "source_label": source_label, "obs_time": observed_at or observed_at_local, diff --git a/web/services/latest_observation_overlay.py b/web/services/latest_observation_overlay.py index 0027b4e8..222546f2 100644 --- a/web/services/latest_observation_overlay.py +++ b/web/services/latest_observation_overlay.py @@ -45,12 +45,19 @@ def _payload_latest_epoch(payload: dict[str, Any], keys: tuple[str, ...]) -> Opt def _block_epoch(block: Any) -> Optional[int]: if not isinstance(block, dict): return None - return _payload_latest_epoch( + canonical_epoch = _payload_latest_epoch( block, ( "observed_at", "observation_time", "obs_time", + ), + ) + if canonical_epoch is not None: + return canonical_epoch + return _payload_latest_epoch( + block, + ( "observed_at_local", "observation_time_local", ), diff --git a/web/services/scan_api.py b/web/services/scan_api.py index e131e8da..c51f0057 100644 --- a/web/services/scan_api.py +++ b/web/services/scan_api.py @@ -23,6 +23,17 @@ def _supports_timing_recorder(func: Any) -> bool: ) +def _supports_keyword(func: Any, name: str) -> bool: + try: + params = signature(func).parameters.values() + except (TypeError, ValueError): + return True + return any( + param.name == name or param.kind == Parameter.VAR_KEYWORD + for param in params + ) + + async def get_scan_terminal_payload( request: Request, *, @@ -36,6 +47,8 @@ async def get_scan_terminal_payload( time_range: str = "today", limit: int = 25, force_refresh: bool = False, + diff: bool = False, + since_snapshot_id: str | None = None, region: str = "", timezone_offset_seconds: int | None = None, ) -> Dict[str, Any]: @@ -67,6 +80,10 @@ async def get_scan_terminal_payload( async def build_payload(): builder = legacy_routes.build_scan_terminal_payload kwargs: Dict[str, Any] = {"force_refresh": force_refresh} + if _supports_keyword(builder, "diff"): + kwargs["diff"] = diff + if _supports_keyword(builder, "since_snapshot_id"): + kwargs["since_snapshot_id"] = since_snapshot_id if _supports_timing_recorder(builder): kwargs["timing_recorder"] = timer return await run_in_threadpool(builder, filters, **kwargs)