diff --git a/.env.example b/.env.example index beb8b63f..5f4d3354 100644 --- a/.env.example +++ b/.env.example @@ -127,6 +127,16 @@ GROQ_API_KEY= POLYWEATHER_GROQ_COMMENTARY_MODEL=openai/gpt-oss-20b POLYWEATHER_GROQ_COMMENTARY_TIMEOUT_SEC=8 POLYWEATHER_GROQ_COMMENTARY_CACHE_TTL_SEC=1800 + +# Optional DeepSeek V4-Flash market scan review for Pro users +POLYWEATHER_SCAN_AI_ENABLED=false +POLYWEATHER_DEEPSEEK_API_KEY= +POLYWEATHER_DEEPSEEK_BASE_URL=https://api.deepseek.com +POLYWEATHER_SCAN_AI_MODEL=deepseek-v4-flash +POLYWEATHER_SCAN_AI_TIMEOUT_SEC=12 +POLYWEATHER_SCAN_AI_CACHE_TTL_SEC=600 +POLYWEATHER_SCAN_AI_MAX_ROWS=40 +POLYWEATHER_SCAN_AI_PROXY_TIMEOUT_MS=18000 POLYWEATHER_PREWARM_CITIES=ankara,istanbul,shanghai,beijing,shenzhen,guangzhou,wuhan,chengdu,chongqing,hong kong,taipei,singapore,tokyo,seoul,busan,london,paris,madrid # Weekly reward / leaderboard diff --git a/frontend/app/api/scan/terminal/ai/route.ts b/frontend/app/api/scan/terminal/ai/route.ts new file mode 100644 index 00000000..8dfe39ad --- /dev/null +++ b/frontend/app/api/scan/terminal/ai/route.ts @@ -0,0 +1,76 @@ +import { NextRequest, NextResponse } from "next/server"; +import { + applyAuthResponseCookies, + buildBackendRequestHeaders, +} from "@/lib/backend-auth"; + +const API_BASE = process.env.POLYWEATHER_API_BASE_URL; +const SCAN_AI_PROXY_TIMEOUT_MS = Number( + process.env.POLYWEATHER_SCAN_AI_PROXY_TIMEOUT_MS || "18000", +); + +export const dynamic = "force-dynamic"; +export const maxDuration = 20; + +export async function POST(req: NextRequest) { + if (!API_BASE) { + return NextResponse.json( + { error: "POLYWEATHER_API_BASE_URL is not configured" }, + { status: 500 }, + ); + } + + let body: unknown = {}; + try { + body = await req.json(); + } catch { + body = {}; + } + + let auth: Awaited> | null = null; + const controller = new AbortController(); + const timeoutId = setTimeout(() => controller.abort(), SCAN_AI_PROXY_TIMEOUT_MS); + + try { + auth = await buildBackendRequestHeaders(req); + const headers = new Headers(auth.headers); + headers.set("Content-Type", "application/json"); + headers.set("Accept", "application/json"); + const res = await fetch(`${API_BASE}/api/scan/terminal/ai`, { + method: "POST", + headers, + cache: "no-store", + signal: controller.signal, + body: JSON.stringify(body || {}), + }); + if (!res.ok) { + const raw = await res.text(); + const response = NextResponse.json( + { error: `Backend returned ${res.status}`, detail: raw.slice(0, 300) }, + { status: res.status === 402 || res.status === 403 ? res.status : 502 }, + ); + return applyAuthResponseCookies(response, auth.response); + } + const data = await res.json(); + const response = NextResponse.json(data, { + headers: { + "Cache-Control": "no-store", + }, + }); + return applyAuthResponseCookies(response, auth.response); + } catch (error) { + const timedOut = controller.signal.aborted; + const response = NextResponse.json( + { + error: timedOut + ? "Scan AI request timed out" + : "Failed to fetch scan AI data", + detail: String(error), + }, + { status: timedOut ? 504 : 500 }, + ); + return auth ? applyAuthResponseCookies(response, auth.response) : response; + } finally { + clearTimeout(timeoutId); + } +} diff --git a/frontend/components/dashboard/Dashboard.module.css b/frontend/components/dashboard/Dashboard.module.css index 2018c28e..9a62691c 100644 --- a/frontend/components/dashboard/Dashboard.module.css +++ b/frontend/components/dashboard/Dashboard.module.css @@ -9391,6 +9391,31 @@ opacity: 0.68; } +.root :global(.scan-ai-button) { + display: inline-flex; + align-items: center; + gap: 8px; + min-height: 42px; + padding: 11px 14px; + border: 1px solid rgba(94, 234, 212, 0.3); + border-radius: 12px; + background: linear-gradient(180deg, rgba(20, 184, 166, 0.22), rgba(14, 116, 144, 0.18)); + color: #a7fff2; + font-size: 13px; + font-weight: 900; + cursor: pointer; +} + +.root :global(.scan-ai-button:hover:not(:disabled)) { + border-color: rgba(94, 234, 212, 0.5); + background: linear-gradient(180deg, rgba(20, 184, 166, 0.32), rgba(14, 116, 144, 0.24)); +} + +.root :global(.scan-ai-button:disabled) { + cursor: not-allowed; + opacity: 0.55; +} + .root :global(.scan-cta-ghost) { color: #27ea98; border-color: rgba(23, 217, 139, 0.3); @@ -9561,6 +9586,12 @@ color: #ffc765; } +.root :global(.scan-status-chip.ai) { + border-color: rgba(94, 234, 212, 0.28); + background: rgba(20, 184, 166, 0.12); + color: #87fff1; +} + .root :global(.scan-table-shell) { display: flex; flex-direction: column; @@ -9759,6 +9790,18 @@ box-shadow: inset 3px 0 0 rgba(23, 217, 139, 0.82); } +.root :global(.scan-opportunity-item.ai-approve) { + border-color: rgba(94, 234, 212, 0.28); +} + +.root :global(.scan-opportunity-item.ai-veto) { + opacity: 0.72; +} + +.root :global(.scan-opportunity-item.ai-veto .scan-opportunity-action) { + color: #94a3b8; +} + .root :global(.scan-opportunity-branch) { position: relative; align-self: stretch; @@ -9857,6 +9900,43 @@ color: #ff6868; } +.root :global(.scan-opportunity-ai) { + grid-column: 2 / -1; + display: grid; + gap: 3px; + min-width: 0; + padding: 8px 10px; + border-radius: 10px; + border: 1px solid rgba(90, 123, 166, 0.12); + background: rgba(5, 14, 25, 0.5); +} + +.root :global(.scan-opportunity-ai b) { + font-size: 11px; + font-weight: 900; +} + +.root :global(.scan-opportunity-ai small) { + overflow: hidden; + color: #8aa1bf; + font-size: 12px; + line-height: 1.45; + text-overflow: ellipsis; + white-space: nowrap; +} + +.root :global(.scan-opportunity-ai.approve b) { + color: #7dfbe7; +} + +.root :global(.scan-opportunity-ai.downgrade b) { + color: #ffc765; +} + +.root :global(.scan-opportunity-ai.veto b) { + color: #ff8585; +} + .root :global(.scan-table-body::-webkit-scrollbar), .root :global(.scan-calendar-view::-webkit-scrollbar), .root :global(.scan-detail-panel::-webkit-scrollbar) { @@ -10875,6 +10955,21 @@ background: linear-gradient(180deg, #25d990, #10b870); } +.root :global(.scan-terminal.light .scan-ai-button) { + color: #075f5a; + border-color: rgba(10, 160, 150, 0.28); + background: linear-gradient(180deg, rgba(214, 250, 246, 0.94), rgba(231, 247, 252, 0.96)); +} + +.root :global(.scan-terminal.light .scan-opportunity-ai) { + border-color: rgba(35, 72, 118, 0.1); + background: rgba(248, 252, 255, 0.76); +} + +.root :global(.scan-terminal.light .scan-opportunity-ai small) { + color: #647a98; +} + .root :global(.scan-terminal.light .scan-cta-ghost), .root :global(.scan-terminal.light .scan-detail-analysis-button) { border-color: rgba(7, 160, 100, 0.24); diff --git a/frontend/components/dashboard/OpportunityTable.tsx b/frontend/components/dashboard/OpportunityTable.tsx index b8105451..f4b93f34 100644 --- a/frontend/components/dashboard/OpportunityTable.tsx +++ b/frontend/components/dashboard/OpportunityTable.tsx @@ -120,6 +120,35 @@ function formatQuoteCents(value?: number | null) { return `${text.replace(/\.0$/, "")}¢`; } +function getAiMeta(row: ScanOpportunityRow, locale: string) { + const decision = String(row.ai_decision || "").toLowerCase(); + if (decision === "veto") { + return { + label: locale === "en-US" ? "AI veto" : "AI 排除", + tone: "veto", + reason: locale === "en-US" ? row.ai_reason_en || row.ai_reason_zh : row.ai_reason_zh || row.ai_reason_en, + }; + } + if (decision === "downgrade") { + return { + label: locale === "en-US" ? "AI downgrade" : "AI 降级", + tone: "downgrade", + reason: locale === "en-US" ? row.ai_reason_en || row.ai_reason_zh : row.ai_reason_zh || row.ai_reason_en, + }; + } + if (row.ai_rank != null || decision === "approve") { + return { + label: locale === "en-US" ? `AI pick ${row.ai_rank || ""}`.trim() : `AI 推荐 ${row.ai_rank || ""}`.trim(), + tone: "approve", + reason: + (locale === "en-US" ? row.ai_reason_en || row.ai_reason_zh : row.ai_reason_zh || row.ai_reason_en) || + row.ai_model_cluster_note || + null, + }; + } + return null; +} + function getDistributionPreview(row: ScanOpportunityRow, tempSymbol?: string | null) { const preview = Array.isArray(row.distribution_preview) ? row.distribution_preview.filter( @@ -349,11 +378,12 @@ export const OpportunityTable = React.memo(function OpportunityTable({ : "EMOS"; const priceLabel = side === "no" ? "NO" : isEn ? "Market" : "市场"; const edgePositive = Number(row.edge_percent || 0) >= 0; + const aiMeta = getAiMeta(row, locale); return ( ); })} diff --git a/frontend/components/dashboard/ScanTerminalDashboard.tsx b/frontend/components/dashboard/ScanTerminalDashboard.tsx index 650a48c1..affcb7a5 100644 --- a/frontend/components/dashboard/ScanTerminalDashboard.tsx +++ b/frontend/components/dashboard/ScanTerminalDashboard.tsx @@ -6,6 +6,7 @@ import { RefreshCw, Moon, Search, + Sparkles, Sun, UserRound, } from "lucide-react"; @@ -252,6 +253,8 @@ function ScanTerminalScreen() { const [activeFilters, setActiveFilters] = useState(DEFAULT_FILTERS); const [terminalData, setTerminalData] = useState(null); const [loading, setLoading] = useState(false); + const [aiLoading, setAiLoading] = useState(false); + const [aiError, setAiError] = useState(null); const [error, setError] = useState(null); const [selectedRowId, setSelectedRowId] = useState(null); const [activeView, setActiveView] = useState("map"); @@ -365,7 +368,9 @@ function ScanTerminalScreen() { ]); const fetchTerminal = async (filters: FilterState, force = false) => { + if (!store.proAccess.subscriptionActive) return; setLoading(true); + setAiError(null); try { const response = await dashboardClient.getScanTerminal(filters, { force }); startTransition(() => { @@ -386,16 +391,45 @@ function ScanTerminalScreen() { } }; - useEffect(() => { - void fetchTerminal(DEFAULT_FILTERS, false); - }, []); + const runAiReview = async () => { + if (!terminalData?.rows.length || aiLoading) return; + setAiLoading(true); + setAiError(null); + try { + const response = await dashboardClient.reviewScanTerminalWithAi({ + filters: terminalData.filters || activeFilters, + snapshotId: terminalData.snapshot_id || null, + }); + startTransition(() => { + setTerminalData(response); + setActiveFilters(response.filters || activeFilters); + setError(response.status === "failed" ? response.stale_reason || null : null); + setSelectedRowId((current) => { + if (current && response.rows.some((row) => row.id === current)) { + return current; + } + return sortRowsByUserTime(response.rows)[0]?.id || response.top_signal?.id || null; + }); + }); + } catch (reviewError) { + setAiError(String(reviewError)); + } finally { + setAiLoading(false); + } + }; useEffect(() => { + if (!store.proAccess.subscriptionActive) return; + void fetchTerminal(DEFAULT_FILTERS, false); + }, [store.proAccess.subscriptionActive]); + + useEffect(() => { + if (!store.proAccess.subscriptionActive) return; const intervalId = window.setInterval(() => { void fetchTerminal(activeFilters, false); }, SCAN_AUTO_REFRESH_MS); return () => window.clearInterval(intervalId); - }, [activeFilters]); + }, [activeFilters, store.proAccess.subscriptionActive]); useEffect(() => { setUserLocalTime(formatUserLocalTime()); @@ -425,6 +459,28 @@ function ScanTerminalScreen() { const scanStatus = terminalData?.status || (loading ? "loading" : error ? "failed" : "ready"); const staleReason = terminalData?.stale_reason || error || null; + const aiScan = terminalData?.ai_scan || null; + const aiStatusText = aiLoading + ? isEn + ? "V4 reviewing" + : "V4 复核中" + : aiError + ? isEn + ? "V4 failed" + : "V4 失败" + : aiScan?.status === "ready" + ? isEn + ? aiScan.cached + ? "V4 cached" + : "V4 reviewed" + : aiScan.cached + ? "V4 缓存" + : "V4 已复核" + : aiScan?.status + ? isEn + ? "Rule scan" + : "使用规则扫描" + : null; useEffect(() => { if (!activeDetailRow) return; @@ -487,6 +543,70 @@ function ScanTerminalScreen() { ); }; + if (store.proAccess.loading) { + return ( +
+
+
+
+
{isEn ? "Checking access" : "正在检查权限"}
+
+ {isEn ? "Preparing your market scan terminal." : "正在准备市场扫描台。"} +
+
+
+
+
+ ); + } + + if (!store.proAccess.subscriptionActive) { + return ( +
+
+
+
+
+ {isEn ? "Market Scan Terminal" : "市场扫描台"} + {isEn ? "Pro access required" : "需要 Pro 权限"} +
+
+ + + + {store.proAccess.authenticated + ? isEn ? "Upgrade Pro" : "升级 Pro" + : isEn ? "Sign in" : "登录"} + +
+
+
+
+
+ {isEn ? "Market scan is a Pro feature" : "市场扫描是 Pro 功能"} +
+
+ {isEn + ? "Free accounts do not trigger rule scans or V4-Flash scans." + : "免费用户不会触发规则扫描，也不会消耗 V4-Flash 扫描资源。"} +
+
+
+
+
+
+ ); + } + return (
@@ -537,6 +657,22 @@ function ScanTerminalScreen() { ? "Start Scan" : "开始扫描"} +
diff --git a/frontend/lib/dashboard-client.ts b/frontend/lib/dashboard-client.ts index 7facfb0c..d6781a12 100644 --- a/frontend/lib/dashboard-client.ts +++ b/frontend/lib/dashboard-client.ts @@ -77,6 +77,7 @@ const pendingCityMarketScanRequests = new Map< >(); const pendingAssistantRequests = new Map>(); const pendingScanTerminalRequests = new Map>(); +const pendingScanTerminalAiRequests = new Map>(); const PRIORITY_WARM_SESSION_KEY = "polyWeather_priority_warm_v1"; type CityCacheMeta = { @@ -438,6 +439,44 @@ export const dashboardClient = { return request; }, + async reviewScanTerminalWithAi(payload: { + filters: ScanTerminalFilters; + snapshotId?: string | null; + }) { + const snapshotId = String(payload.snapshotId || "").trim(); + const requestKey = [ + snapshotId || "latest", + JSON.stringify(payload.filters || {}), + ].join("::"); + const existing = pendingScanTerminalAiRequests.get(requestKey); + if (existing) { + return existing; + } + const request = fetch("/api/scan/terminal/ai", { + method: "POST", + headers: { + Accept: "application/json", + "Content-Type": "application/json", + }, + cache: "no-store", + body: JSON.stringify({ + filters: payload.filters, + snapshot_id: snapshotId || null, + }), + }) + .then(async (response) => { + if (!response.ok) { + throw new Error(`HTTP ${response.status}`); + } + return response.json() as Promise; + }) + .finally(() => { + pendingScanTerminalAiRequests.delete(requestKey); + }); + pendingScanTerminalAiRequests.set(requestKey, request); + return request; + }, + async getHistory(cityName: string, options?: { includeRecords?: boolean }) { const includeRecords = options?.includeRecords === true; const requestKey = `${normalizeCityName(cityName)}::${ diff --git a/frontend/lib/dashboard-types.ts b/frontend/lib/dashboard-types.ts index 57dbd3c7..d6e6e4c9 100644 --- a/frontend/lib/dashboard-types.ts +++ b/frontend/lib/dashboard-types.ts @@ -535,10 +535,29 @@ export interface ScanOpportunityRow { signal_status?: string | null; candidate_count?: number | null; resolved_market_type?: string | null; + ai_decision?: "approve" | "veto" | "downgrade" | "neutral" | string | null; + ai_rank?: number | null; + ai_confidence?: string | null; + ai_reason_zh?: string | null; + ai_reason_en?: string | null; + ai_model_cluster_note?: string | null; } export interface PrimarySignal extends ScanOpportunityRow {} +export interface ScanAiReview { + status?: "ready" | "disabled" | "missing_key" | "failed" | "no_rows" | "no_snapshot" | "snapshot_mismatch" | string; + model?: string | null; + cached?: boolean; + generated_at?: string | null; + reason?: string | null; + summary_zh?: string | null; + summary_en?: string | null; + recommended_count?: number | null; + vetoed_count?: number | null; + downgraded_count?: number | null; +} + export interface ScanTerminalResponse { generated_at: string; snapshot_id?: string | null; @@ -563,6 +582,7 @@ export interface ScanTerminalResponse { }; top_signal?: PrimarySignal | null; rows: ScanOpportunityRow[]; + ai_scan?: ScanAiReview | null; } export interface IntradayMeteorologySignal { diff --git a/web/routes.py b/web/routes.py index a0ac0553..1ad1788b 100644 --- a/web/routes.py +++ b/web/routes.py @@ -32,7 +32,10 @@ from web.analysis_service import ( _build_city_market_scan_payload, _build_city_summary_payload, ) -from web.scan_terminal_service import build_scan_terminal_payload +from web.scan_terminal_service import ( + build_scan_terminal_ai_payload, + build_scan_terminal_payload, +) from web.core import ( AnalyticsEventRequest, CITIES, @@ -1728,3 +1731,21 @@ async def scan_terminal( force_refresh=force_refresh, ) + +@router.post("/api/scan/terminal/ai") +async def scan_terminal_ai(request: Request): + _assert_entitlement(request) + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + raise HTTPException(status_code=400, detail="Invalid JSON body") + filters = body.get("filters") if isinstance(body.get("filters"), dict) else {} + snapshot_id = str(body.get("snapshot_id") or "").strip() or None + return await run_in_threadpool( + build_scan_terminal_ai_payload, + filters, + snapshot_id=snapshot_id, + ) + diff --git a/web/scan_terminal_service.py b/web/scan_terminal_service.py index e1db2a77..7ac9e824 100644 --- a/web/scan_terminal_service.py +++ b/web/scan_terminal_service.py @@ -10,6 +10,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timedelta from typing import Any, Dict, List, Optional +import httpx from loguru import logger from web.analysis_service import _analyze, _build_city_market_scan_payload @@ -18,6 +19,8 @@ from web.core import CITIES _SCAN_TERMINAL_CACHE_LOCK = threading.Lock() _SCAN_TERMINAL_CACHE: Dict[str, Dict[str, Any]] = {} _SCAN_TERMINAL_REFRESHING: set[str] = set() +_SCAN_TERMINAL_AI_CACHE_LOCK = threading.Lock() +_SCAN_TERMINAL_AI_CACHE: Dict[str, Dict[str, Any]] = {} SCAN_TERMINAL_PAYLOAD_TTL_SEC = max( 5, int(os.getenv("POLYWEATHER_SCAN_TERMINAL_PAYLOAD_TTL_SEC", "30")), @@ -26,6 +29,27 @@ SCAN_TERMINAL_BUILD_TIMEOUT_SEC = max( 8, int(os.getenv("POLYWEATHER_SCAN_TERMINAL_BUILD_TIMEOUT_SEC", "22")), ) +SCAN_AI_MODEL = str( + os.getenv("POLYWEATHER_SCAN_AI_MODEL") or "deepseek-v4-flash" +).strip() +SCAN_AI_BASE_URL = str( + os.getenv("POLYWEATHER_DEEPSEEK_BASE_URL") or "https://api.deepseek.com" +).strip().rstrip("/") +SCAN_AI_ENABLED = str( + os.getenv("POLYWEATHER_SCAN_AI_ENABLED") or "false" +).strip().lower() in {"1", "true", "yes", "on"} +SCAN_AI_TIMEOUT_SEC = max( + 3, + int(os.getenv("POLYWEATHER_SCAN_AI_TIMEOUT_SEC", "12")), +) +SCAN_AI_CACHE_TTL_SEC = max( + 30, + int(os.getenv("POLYWEATHER_SCAN_AI_CACHE_TTL_SEC", "600")), +) +SCAN_AI_MAX_ROWS = max( + 1, + int(os.getenv("POLYWEATHER_SCAN_AI_MAX_ROWS", "40")), +) def _safe_float(value: Any) -> Optional[float]: @@ -250,6 +274,291 @@ def _build_failed_scan_terminal_payload( } +def _extract_ai_json_object(raw_text: str) -> Dict[str, Any]: + text = str(raw_text or "").strip() + if not text: + raise ValueError("empty AI content") + try: + parsed = json.loads(text) + if isinstance(parsed, dict): + return parsed + except Exception: + pass + start = text.find("{") + end = text.rfind("}") + if start >= 0 and end > start: + parsed = json.loads(text[start : end + 1]) + if isinstance(parsed, dict): + return parsed + raise ValueError("AI content is not a JSON object") + + +def _scan_ai_cache_key(snapshot_id: str, filters: Dict[str, Any]) -> str: + raw = json.dumps( + { + "snapshot_id": snapshot_id, + "filters": _normalize_scan_terminal_filters(filters), + "model": SCAN_AI_MODEL, + "max_rows": SCAN_AI_MAX_ROWS, + }, + sort_keys=True, + ensure_ascii=False, + ) + return hashlib.sha256(raw.encode("utf-8")).hexdigest() + + +def _get_cached_scan_ai_result(snapshot_id: str, filters: Dict[str, Any]) -> Optional[Dict[str, Any]]: + cache_key = _scan_ai_cache_key(snapshot_id, filters) + now = time.time() + with _SCAN_TERMINAL_AI_CACHE_LOCK: + cached = _SCAN_TERMINAL_AI_CACHE.get(cache_key) + if not cached: + return None + cached_at = float(cached.get("cached_at") or 0.0) + if now - cached_at >= float(SCAN_AI_CACHE_TTL_SEC): + return None + result = cached.get("result") + if isinstance(result, dict): + return dict(result) + return None + + +def _set_cached_scan_ai_result(snapshot_id: str, filters: Dict[str, Any], result: Dict[str, Any]) -> None: + cache_key = _scan_ai_cache_key(snapshot_id, filters) + with _SCAN_TERMINAL_AI_CACHE_LOCK: + _SCAN_TERMINAL_AI_CACHE[cache_key] = { + "cached_at": time.time(), + "result": result, + } + + +def _compact_ai_candidate(row: Dict[str, Any]) -> Dict[str, Any]: + return { + "id": row.get("id"), + "city": row.get("city_display_name") or row.get("display_name") or row.get("city"), + "local_time": row.get("local_time"), + "current_temp": row.get("current_temp"), + "current_max_so_far": row.get("current_max_so_far"), + "deb_prediction": row.get("deb_prediction"), + "action": row.get("action"), + "side": row.get("side"), + "target_label": row.get("target_label"), + "target_value": row.get("target_value"), + "target_threshold": row.get("target_threshold"), + "target_unit": row.get("target_unit"), + "model_probability": row.get("model_probability"), + "market_probability": row.get("market_probability"), + "model_event_probability": row.get("model_event_probability"), + "market_event_probability": row.get("market_event_probability"), + "yes_ask": row.get("yes_ask"), + "no_ask": row.get("no_ask"), + "ask": row.get("ask"), + "spread": row.get("spread"), + "quote_age_ms": row.get("quote_age_ms"), + "edge_percent": row.get("edge_percent"), + "final_score": row.get("final_score"), + "window_phase": row.get("window_phase"), + "remaining_window_minutes": row.get("remaining_window_minutes"), + "distribution_preview": (row.get("distribution_preview") or [])[:6], + "peak_value": row.get("peak_value"), + "peak_probability": row.get("peak_probability"), + "cluster_role": row.get("cluster_role"), + "cluster_core_low": row.get("cluster_core_low"), + "cluster_core_high": row.get("cluster_core_high"), + "cluster_median": row.get("cluster_median"), + "cluster_deb_reference": row.get("cluster_deb_reference"), + "cluster_model_count": row.get("cluster_model_count"), + "trend_alignment": row.get("trend_alignment"), + "tradable": row.get("tradable"), + "accepting_orders": row.get("accepting_orders"), + } + + +def _build_scan_ai_prompt(payload: Dict[str, Any]) -> Dict[str, Any]: + rows = [ + _compact_ai_candidate(row) + for row in (payload.get("rows") or [])[:SCAN_AI_MAX_ROWS] + if isinstance(row, dict) and row.get("id") + ] + return { + "snapshot_id": payload.get("snapshot_id"), + "generated_at": payload.get("generated_at"), + "summary": payload.get("summary") or {}, + "filters": payload.get("filters") or {}, + "candidates": rows, + } + + +def _call_deepseek_scan_ai(ai_input: Dict[str, Any]) -> Dict[str, Any]: + api_key = str(os.getenv("POLYWEATHER_DEEPSEEK_API_KEY") or "").strip() + if not api_key: + raise RuntimeError("POLYWEATHER_DEEPSEEK_API_KEY is not configured") + + system_prompt = ( + "你是 PolyWeather 的付费市场扫描员。你只能基于用户提供的 JSON 快照做判断，" + "不得编造城市、价格、概率、盘口或天气数据。你的任务是从候选 rows 中选择真正值得看的机会，" + "尤其要检查 DEB、EMOS 分布、模型集群、当前实测、峰值窗口和盘口价格是否一致。" + "若模型集群不支持某个 YES，不要机械推荐 YES；可以 veto 或改为偏向 NO 候选。" + "只能引用输入中的 row id。必须输出 JSON object，字段为 recommendations、vetoed、downgraded、summary_zh、summary_en。" + ) + user_payload = { + "task": ( + "Review these existing PolyWeather market candidates. " + "Return strict JSON only. recommendations items need row_id, rank, decision, confidence, " + "reason_zh, reason_en, model_cluster_note. vetoed/downgraded items need row_id and reason." + ), + "snapshot": ai_input, + } + with httpx.Client(timeout=float(SCAN_AI_TIMEOUT_SEC)) as client: + response = client.post( + f"{SCAN_AI_BASE_URL}/chat/completions", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json={ + "model": SCAN_AI_MODEL, + "temperature": 0.1, + "max_tokens": 2200, + "response_format": {"type": "json_object"}, + "messages": [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": json.dumps(user_payload, ensure_ascii=False), + }, + ], + }, + ) + response.raise_for_status() + data = response.json() + content = ( + ((data.get("choices") or [{}])[0].get("message") or {}).get("content") + if isinstance(data, dict) + else None + ) + return _extract_ai_json_object(str(content or "")) + + +def _normalize_ai_items(raw_items: Any) -> List[Dict[str, Any]]: + if not isinstance(raw_items, list): + return [] + out: List[Dict[str, Any]] = [] + for item in raw_items: + if isinstance(item, str): + out.append({"row_id": item}) + elif isinstance(item, dict): + row_id = str(item.get("row_id") or item.get("id") or "").strip() + if row_id: + out.append({**item, "row_id": row_id}) + return out + + +def _merge_scan_ai_result(payload: Dict[str, Any], ai_raw: Dict[str, Any], *, cached: bool = False) -> Dict[str, Any]: + rows = [dict(row) for row in (payload.get("rows") or []) if isinstance(row, dict)] + by_id = {str(row.get("id")): row for row in rows if row.get("id")} + recommendations = _normalize_ai_items(ai_raw.get("recommendations")) + vetoed = _normalize_ai_items(ai_raw.get("vetoed")) + downgraded = _normalize_ai_items(ai_raw.get("downgraded")) + + veto_ids = {str(item.get("row_id")) for item in vetoed} + downgrade_ids = {str(item.get("row_id")) for item in downgraded} + recommended_ids: set[str] = set() + + for item in vetoed: + row = by_id.get(str(item.get("row_id"))) + if not row: + continue + row["ai_decision"] = "veto" + row["ai_reason_zh"] = item.get("reason_zh") or item.get("reason") + row["ai_reason_en"] = item.get("reason_en") + for item in downgraded: + row = by_id.get(str(item.get("row_id"))) + if not row: + continue + row["ai_decision"] = "downgrade" + row["ai_reason_zh"] = item.get("reason_zh") or item.get("reason") + row["ai_reason_en"] = item.get("reason_en") + for fallback_rank, item in enumerate(recommendations, start=1): + row_id = str(item.get("row_id")) + row = by_id.get(row_id) + if not row: + continue + if row_id in veto_ids: + continue + recommended_ids.add(row_id) + row["ai_decision"] = str(item.get("decision") or "approve").strip().lower() or "approve" + row["ai_rank"] = _safe_int(item.get("rank"), fallback_rank) + row["ai_confidence"] = item.get("confidence") + row["ai_reason_zh"] = item.get("reason_zh") or item.get("reason") + row["ai_reason_en"] = item.get("reason_en") + row["ai_model_cluster_note"] = item.get("model_cluster_note") + + for row in rows: + row_id = str(row.get("id")) + if row_id not in recommended_ids and row_id not in veto_ids and row_id not in downgrade_ids: + row["ai_decision"] = row.get("ai_decision") or "neutral" + + def _ai_sort_key(row: Dict[str, Any]) -> tuple: + decision = str(row.get("ai_decision") or "").lower() + if decision == "veto": + tier = 3 + elif decision == "downgrade": + tier = 2 + elif row.get("ai_rank") is not None: + tier = 0 + else: + tier = 1 + return ( + tier, + _safe_int(row.get("ai_rank"), 999), + -float(row.get("final_score") or 0.0), + -float(row.get("edge_percent") or 0.0), + ) + + rows.sort(key=_ai_sort_key) + top_signal = next( + (row for row in rows if str(row.get("ai_decision") or "").lower() != "veto"), + rows[0] if rows else None, + ) + ai_scan = { + "status": "ready", + "model": SCAN_AI_MODEL, + "cached": cached, + "generated_at": datetime.utcnow().isoformat() + "Z", + "summary_zh": ai_raw.get("summary_zh"), + "summary_en": ai_raw.get("summary_en"), + "recommended_count": sum(1 for row in rows if row.get("ai_rank") is not None), + "vetoed_count": sum(1 for row in rows if row.get("ai_decision") == "veto"), + "downgraded_count": sum(1 for row in rows if row.get("ai_decision") == "downgrade"), + } + merged = { + **payload, + "rows": rows, + "top_signal": top_signal, + "ai_scan": ai_scan, + } + return merged + + +def _build_scan_ai_unavailable_payload( + payload: Dict[str, Any], + *, + status: str, + reason: str, +) -> Dict[str, Any]: + return { + **payload, + "ai_scan": { + "status": status, + "model": SCAN_AI_MODEL, + "cached": False, + "generated_at": datetime.utcnow().isoformat() + "Z", + "reason": reason, + }, + } + + def _resolve_time_range_dates(data: Dict[str, Any], time_range: str) -> List[str]: local_date = str(data.get("local_date") or "").strip() multi_model_daily = data.get("multi_model_daily") or {} @@ -610,3 +919,61 @@ def build_scan_terminal_payload( ) return _build_scan_terminal_payload_uncached(filters, force_refresh=force_refresh) + + +def build_scan_terminal_ai_payload( + raw_filters: Optional[Dict[str, Any]] = None, + *, + snapshot_id: Optional[str] = None, +) -> Dict[str, Any]: + filters = _normalize_scan_terminal_filters(raw_filters) + payload = build_scan_terminal_payload(filters, force_refresh=False) + current_snapshot_id = str(payload.get("snapshot_id") or "").strip() + requested_snapshot_id = str(snapshot_id or "").strip() + if requested_snapshot_id and current_snapshot_id and requested_snapshot_id != current_snapshot_id: + return _build_scan_ai_unavailable_payload( + payload, + status="snapshot_mismatch", + reason="scan snapshot changed; refresh the scan before running AI review", + ) + if not current_snapshot_id: + return _build_scan_ai_unavailable_payload( + payload, + status="no_snapshot", + reason="no scan snapshot is available for AI review", + ) + if not payload.get("rows"): + return _build_scan_ai_unavailable_payload( + payload, + status="no_rows", + reason="no candidate rows are available for AI review", + ) + if not SCAN_AI_ENABLED: + return _build_scan_ai_unavailable_payload( + payload, + status="disabled", + reason="POLYWEATHER_SCAN_AI_ENABLED is not enabled", + ) + if not str(os.getenv("POLYWEATHER_DEEPSEEK_API_KEY") or "").strip(): + return _build_scan_ai_unavailable_payload( + payload, + status="missing_key", + reason="POLYWEATHER_DEEPSEEK_API_KEY is not configured", + ) + + cached = _get_cached_scan_ai_result(current_snapshot_id, filters) + if cached is not None: + return _merge_scan_ai_result(payload, cached, cached=True) + + try: + ai_input = _build_scan_ai_prompt(payload) + ai_raw = _call_deepseek_scan_ai(ai_input) + _set_cached_scan_ai_result(current_snapshot_id, filters, ai_raw) + return _merge_scan_ai_result(payload, ai_raw, cached=False) + except Exception as exc: + logger.warning("scan terminal AI review failed: {}", exc) + return _build_scan_ai_unavailable_payload( + payload, + status="failed", + reason=str(exc), + )