From 6c203bee609190f003c2a95863f809074bad2c5b Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Fri, 1 May 2026 11:27:16 +0800 Subject: [PATCH] This version of Antigravity is no longer supported. Please upgrade to receive the latest features. --- .../dashboard/ScanTerminalCard.module.css | 14 +- .../scan-terminal/AiCityTemperatureChart.tsx | 98 +++++++---- scripts/backtest_metar_calibrated_path.py | 157 ++++++++++++++++-- src/database/runtime_state.py | 86 ++++++++++ web/analysis_service.py | 72 ++++++++ 5 files changed, 376 insertions(+), 51 deletions(-) diff --git a/frontend/components/dashboard/ScanTerminalCard.module.css b/frontend/components/dashboard/ScanTerminalCard.module.css index 781c5c35..b6b905c5 100644 --- a/frontend/components/dashboard/ScanTerminalCard.module.css +++ b/frontend/components/dashboard/ScanTerminalCard.module.css @@ -994,7 +994,19 @@ width: 18px; height: 3px; border-radius: 999px; - background: #4da3ff; + background: rgba(100, 116, 139, 0.72); +} + +.root :global(.scan-ai-city-chart-legend i.forecast) { + background: repeating-linear-gradient( + 90deg, + rgba(100, 116, 139, 0.72) 0 6px, + transparent 6px 10px + ); +} + +.root :global(.scan-ai-city-chart-legend i.calibrated) { + background: #38bdf8; } .root :global(.scan-ai-city-chart-legend i.observation) { diff --git a/frontend/components/dashboard/scan-terminal/AiCityTemperatureChart.tsx b/frontend/components/dashboard/scan-terminal/AiCityTemperatureChart.tsx index 854fef92..410d5eca 100644 --- a/frontend/components/dashboard/scan-terminal/AiCityTemperatureChart.tsx +++ b/frontend/components/dashboard/scan-terminal/AiCityTemperatureChart.tsx @@ -86,16 +86,17 @@ export function AiCityTemperatureChart({ detail }: { detail: CityDetail }) { (lastChartDataRef.current?.cityKey === cityKey ? lastChartDataRef.current.data : null); - const forecastLabel = chartData?.datasets.hasMgmHourly - ? locale === "en-US" - ? "MGM forecast" - : "MGM 预测" - : locale === "en-US" - ? "DEB forecast" - : "DEB 预测"; + const forecastLabel = locale === "en-US" ? "DEB baseline" : "DEB 原始路径"; + const calibratedLabel = + locale === "en-US" + ? "METAR-calibrated path" + : "METAR 修正路径"; const observationLabel = chartData?.observationLabel || (locale === "en-US" ? "METAR obs" : "METAR 实况"); + const hasCalibratedPath = Boolean( + chartData?.datasets.calibratedFuture.some((value) => value != null), + ); const canvasRef = useChart(() => { if (!chartData) { return { @@ -103,36 +104,53 @@ export function AiCityTemperatureChart({ detail }: { detail: CityDetail }) { type: "line", } satisfies ChartConfiguration<"line">; } - const forecastPoints = chartData.datasets.hasMgmHourly - ? chartData.datasets.mgmHourlyPoints - : chartData.datasets.debPast.map( + const datasets: NonNullable< + ChartConfiguration<"line">["data"] + >["datasets"] = [ + { + borderColor: "rgba(100, 116, 139, 0.72)", + borderDash: [6, 4], + borderWidth: 1.6, + data: chartData.datasets.debPast.map( (value, index) => value ?? chartData.datasets.debFuture[index], - ); + ), + fill: false, + label: forecastLabel, + pointRadius: 0, + spanGaps: true, + tension: 0.28, + }, + ]; + + if (hasCalibratedPath) { + datasets.push({ + borderColor: "#38bdf8", + borderWidth: 2.3, + data: chartData.datasets.calibratedFuture, + fill: false, + label: calibratedLabel, + pointHoverRadius: 5, + pointRadius: 0, + spanGaps: true, + tension: 0.32, + }); + } + + datasets.push({ + backgroundColor: "#22C55E", + borderColor: "#22C55E", + borderWidth: 0, + data: chartData.datasets.metarPoints, + fill: false, + label: observationLabel, + pointHoverRadius: 5, + pointRadius: 3.5, + showLine: false, + }); + return { data: { - datasets: [ - { - borderColor: "#4DA3FF", - borderWidth: 2, - data: forecastPoints, - fill: false, - label: forecastLabel, - pointRadius: 0, - spanGaps: true, - tension: 0.32, - }, - { - backgroundColor: "#22C55E", - borderColor: "#22C55E", - borderWidth: 0, - data: chartData.datasets.metarPoints, - fill: false, - label: observationLabel, - pointHoverRadius: 5, - pointRadius: 3.5, - showLine: false, - }, - ], + datasets, labels: chartData.times, }, options: { @@ -186,7 +204,14 @@ export function AiCityTemperatureChart({ detail }: { detail: CityDetail }) { }, type: "line", } satisfies ChartConfiguration<"line">; - }, [chartData, detail.temp_symbol, forecastLabel, observationLabel]); + }, [ + calibratedLabel, + chartData, + detail.temp_symbol, + forecastLabel, + hasCalibratedPath, + observationLabel, + ]); useEffect(() => { if (shouldRenderChart) return; @@ -226,6 +251,9 @@ export function AiCityTemperatureChart({ detail }: { detail: CityDetail }) { {chartData ? (
{forecastLabel} + {hasCalibratedPath ? ( + {calibratedLabel} + ) : null} {observationLabel}
) : null} diff --git a/scripts/backtest_metar_calibrated_path.py b/scripts/backtest_metar_calibrated_path.py index d779cdaf..d388c9a5 100644 --- a/scripts/backtest_metar_calibrated_path.py +++ b/scripts/backtest_metar_calibrated_path.py @@ -358,6 +358,114 @@ def evaluate_city_date(con: sqlite3.Connection, city: str, date: str, min_obs: i return results +def load_path_snapshot_rows( + con: sqlite3.Connection, + *, + cities: list[str], + dates_filter: set[str], +) -> list[dict[str, Any]]: + try: + con.execute("select 1 from intraday_path_snapshots_store limit 1").fetchone() + except sqlite3.Error: + return [] + params: list[Any] = [] + clauses: list[str] = [] + if cities: + clauses.append("city in (" + ",".join("?" for _ in cities) + ")") + params.extend(cities) + if dates_filter: + clauses.append("target_date in (" + ",".join("?" for _ in dates_filter) + ")") + params.extend(sorted(dates_filter)) + where = " where " + " and ".join(clauses) if clauses else "" + rows = con.execute( + f"select payload_json from intraday_path_snapshots_store{where} order by id asc", + params, + ).fetchall() + out: list[dict[str, Any]] = [] + for row in rows: + try: + payload = json.loads(row["payload_json"]) + except Exception: + continue + if isinstance(payload, dict): + out.append(payload) + return out + + +def observation_rows_from_snapshot(snapshot: dict[str, Any]) -> list[Observation]: + rows: list[Observation] = [] + for key in ("metar_today_obs", "settlement_today_obs"): + raw_rows = snapshot.get(key) + if not isinstance(raw_rows, list): + continue + for item in raw_rows: + if not isinstance(item, dict): + continue + temp = sf(item.get("temp")) + time_text = str(item.get("time") or "").strip()[:5] + if temp is not None and hm_to_minutes(time_text) is not None: + rows.append(Observation(time_text, temp)) + return dedupe_observations(rows) + + +def evaluate_path_snapshot( + con: sqlite3.Connection, + snapshot: dict[str, Any], + min_obs: int, +) -> SampleResult | None: + city = str(snapshot.get("city") or "").strip().lower() + date = str(snapshot.get("target_date") or snapshot.get("date") or "").strip() + if not city or not date: + return None + actual_high = get_actual_high(con, city, date) + deb_high = sf(snapshot.get("deb_prediction")) + if actual_high is None or deb_high is None: + return None + path = snapshot.get("deb_base_path") or {} + times = path.get("times") if isinstance(path, dict) else [] + deb_path = path.get("temps") if isinstance(path, dict) else [] + if not isinstance(times, list) or not isinstance(deb_path, list) or not times: + return None + deb_values = [sf(v) for v in deb_path] + observations = observation_rows_from_snapshot(snapshot) + if len(observations) < min_obs: + return None + local_time = str(snapshot.get("local_time") or "").strip() + current_minute = hm_to_minutes(local_time) + if current_minute is None: + current_minute = hm_to_minutes(observations[-1].time) + if current_minute is None: + return None + forecast = snapshot.get("forecast") if isinstance(snapshot.get("forecast"), dict) else {} + reversion_minute = hm_to_minutes(forecast.get("sunset")) or hm_to_minutes("18:00") or 18 * 60 + calibrated_path, adjustment = calibrated_future_path( + times=[str(t) for t in times], + deb_path=deb_values, + observations=observations, + current_minute=current_minute, + reversion_minute=reversion_minute, + ) + future_values = [v for v in calibrated_path if v is not None] + current = snapshot.get("current") if isinstance(snapshot.get("current"), dict) else {} + max_so_far = sf(current.get("max_so_far")) + observed_so_far = max([o.temp for o in observations] + ([max_so_far] if max_so_far is not None else [])) + calibrated_high = max([observed_so_far, *future_values], default=observed_so_far) + return SampleResult( + city=city, + date=date, + current_time=local_time or observations[-1].time, + obs_count=len(observations), + actual_high=actual_high, + deb_high=deb_high, + calibrated_high=calibrated_high, + deb_abs_error=abs(deb_high - actual_high), + calibrated_abs_error=abs(calibrated_high - actual_high), + delta_vs_deb=adjustment if adjustment is not None else 0.0, + bucket_deb_hit=bucket_hit(city, deb_high, actual_high), + bucket_calibrated_hit=bucket_hit(city, calibrated_high, actual_high), + ) + + def summarize(samples: list[SampleResult]) -> dict[str, Any]: if not samples: return {"samples": 0} @@ -396,6 +504,12 @@ def main() -> int: parser.add_argument("--city", action="append", help="City key; can be repeated. Defaults to cities found in daily_records_store.") parser.add_argument("--date", action="append", help="YYYY-MM-DD; can be repeated.") parser.add_argument("--min-obs", type=int, default=2) + parser.add_argument( + "--source", + choices=("strict", "snapshots", "both"), + default="both", + help="strict uses reconstructed legacy stores; snapshots uses intraday_path_snapshots_store.", + ) parser.add_argument("--output", default=str(ROOT / "tmp_metar_calibration_backtest.csv")) args = parser.parse_args() @@ -406,28 +520,41 @@ def main() -> int: dates_filter = set(args.date or []) all_samples: list[SampleResult] = [] - skipped = {"no_records_or_inputs": 0} - for city in cities: - rows = con.execute( - "select distinct target_date from daily_records_store where city=? order by target_date", - (city,), - ).fetchall() - for row in rows: - date = row[0] - if dates_filter and date not in dates_filter: - continue - samples = evaluate_city_date(con, city, date, args.min_obs) - if samples: - all_samples.extend(samples) + skipped = {"no_records_or_inputs": 0, "snapshots_unusable": 0} + if args.source in {"strict", "both"}: + for city in cities: + rows = con.execute( + "select distinct target_date from daily_records_store where city=? order by target_date", + (city,), + ).fetchall() + for row in rows: + date = row[0] + if dates_filter and date not in dates_filter: + continue + samples = evaluate_city_date(con, city, date, args.min_obs) + if samples: + all_samples.extend(samples) + else: + skipped["no_records_or_inputs"] += 1 + if args.source in {"snapshots", "both"}: + snapshot_rows = load_path_snapshot_rows( + con, + cities=cities or [], + dates_filter=dates_filter, + ) + for snapshot in snapshot_rows: + sample = evaluate_path_snapshot(con, snapshot, args.min_obs) + if sample: + all_samples.append(sample) else: - skipped["no_records_or_inputs"] += 1 + skipped["snapshots_unusable"] += 1 summary = summarize(all_samples) write_csv(Path(args.output), all_samples) print(json.dumps({"summary": summary, "skipped": skipped, "output": args.output}, ensure_ascii=False, indent=2)) if not all_samples: print( - "No usable samples. Need matching daily_records + open_meteo_cache hourly forecast + intraday observations for the same city/date.", + "No usable samples. Need strict store matches or rows in intraday_path_snapshots_store with later actual_high.", file=sys.stderr, ) return 2 diff --git a/src/database/runtime_state.py b/src/database/runtime_state.py index 033c4da0..a7109374 100644 --- a/src/database/runtime_state.py +++ b/src/database/runtime_state.py @@ -209,6 +209,26 @@ class RuntimeStateDB: conn.execute( "CREATE INDEX IF NOT EXISTS idx_official_intraday_obs_station_date ON official_intraday_observations_store(source_code, station_code, target_date, observation_time)" ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS intraday_path_snapshots_store ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + city TEXT NOT NULL, + target_date TEXT NOT NULL, + snapshot_time TEXT NOT NULL, + local_time TEXT, + deb_prediction REAL, + forecast_today_high REAL, + current_temp REAL, + max_so_far REAL, + observation_count INTEGER NOT NULL DEFAULT 0, + payload_json TEXT NOT NULL + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_intraday_path_snapshots_city_date ON intraday_path_snapshots_store(city, target_date, id DESC)" + ) conn.commit() @@ -1053,6 +1073,72 @@ class OfficialIntradayObservationRepository: return out +class IntradayPathSnapshotRepository: + def __init__(self, db: Optional[RuntimeStateDB] = None): + self.db = db or RuntimeStateDB.instance() + + def append_snapshot(self, payload: Dict[str, Any]) -> None: + observations = [] + for key in ("metar_today_obs", "settlement_today_obs"): + rows = payload.get(key) + if isinstance(rows, list): + observations.extend(rows) + with self.db.connect() as conn: + conn.execute( + """ + INSERT INTO intraday_path_snapshots_store ( + city, target_date, snapshot_time, local_time, + deb_prediction, forecast_today_high, current_temp, + max_so_far, observation_count, payload_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + payload.get("city"), + payload.get("target_date"), + payload.get("snapshot_time"), + payload.get("local_time"), + payload.get("deb_prediction"), + payload.get("forecast_today_high"), + payload.get("current_temp"), + payload.get("max_so_far"), + len(observations), + json.dumps(payload, ensure_ascii=False), + ), + ) + conn.commit() + + def load_rows_by_city_date(self, city: str, target_date: str) -> List[Dict[str, Any]]: + with self.db.connect() as conn: + rows = conn.execute( + """ + SELECT payload_json + FROM intraday_path_snapshots_store + WHERE city = ? AND target_date = ? + ORDER BY id ASC + """, + (city, target_date), + ).fetchall() + out: List[Dict[str, Any]] = [] + for row in rows: + try: + out.append(json.loads(row["payload_json"])) + except Exception: + continue + return out + + def load_all_rows(self) -> List[Dict[str, Any]]: + with self.db.connect() as conn: + rows = conn.execute( + "SELECT payload_json FROM intraday_path_snapshots_store ORDER BY id ASC" + ).fetchall() + out: List[Dict[str, Any]] = [] + for row in rows: + try: + out.append(json.loads(row["payload_json"])) + except Exception: + continue + return out + def _top_bucket(snapshot: Optional[List[Dict[str, Any]]]) -> Optional[int]: best_value = None best_prob = -1.0 diff --git a/web/analysis_service.py b/web/analysis_service.py index 4b9d12ef..ae015891 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -32,6 +32,7 @@ from src.data_collection.country_networks import build_country_network_snapshot from src.data_collection.city_registry import ALIASES, CITY_REGISTRY from src.data_collection.city_time import get_city_utc_offset_seconds from src.data_collection.nmc_sources import NMC_CITY_REFERENCES +from src.database.runtime_state import IntradayPathSnapshotRepository from src.models.lgbm_daily_high import predict_lgbm_daily_high TURKISH_MGM_CITIES = {"ankara", "istanbul"} @@ -1539,6 +1540,75 @@ def _build_intraday_meteorology(data: Dict[str, Any]) -> Dict[str, Any]: } +def _archive_intraday_path_snapshot(city: str, result: Dict[str, Any]) -> None: + """Persist replayable intraday path inputs visible at analysis time.""" + hourly = result.get("hourly") or {} + times = hourly.get("times") if isinstance(hourly, dict) else [] + temps = hourly.get("temps") if isinstance(hourly, dict) else [] + if not isinstance(times, list) or not isinstance(temps, list) or not times: + return + + forecast = result.get("forecast") or {} + deb = result.get("deb") or {} + current = result.get("current") or {} + forecast_today_high = _sf(forecast.get("today_high")) + deb_prediction = _sf(deb.get("prediction")) + offset = ( + deb_prediction - forecast_today_high + if deb_prediction is not None and forecast_today_high is not None + else 0.0 + ) + deb_base_temps = [ + round(float(value) + offset, 1) if _sf(value) is not None else None + for value in temps + ] + utc_offset = int(result.get("utc_offset_seconds") or 0) + snapshot_time = datetime.now(timezone.utc).astimezone( + timezone(timedelta(seconds=utc_offset)) + ).isoformat(timespec="seconds") + payload = { + "schema_version": 1, + "city": city, + "target_date": str(result.get("local_date") or "").strip(), + "snapshot_time": snapshot_time, + "local_time": str(result.get("local_time") or "").strip(), + "utc_offset_seconds": utc_offset, + "temp_symbol": result.get("temp_symbol"), + "deb_prediction": deb_prediction, + "forecast_today_high": forecast_today_high, + "deb_base_path": { + "times": [str(item) for item in times], + "temps": deb_base_temps, + "source": "hourly_plus_deb_offset", + "offset": round(offset, 3), + }, + "hourly": { + "times": [str(item) for item in times], + "temps": temps, + }, + "metar_today_obs": result.get("metar_today_obs") or [], + "settlement_today_obs": result.get("settlement_today_obs") or [], + "current": { + "temp": _sf(current.get("temp")), + "max_so_far": _sf(current.get("max_so_far")), + "obs_time": current.get("obs_time"), + "settlement_source": current.get("settlement_source"), + "settlement_source_label": current.get("settlement_source_label"), + }, + "forecast": { + "today_high": forecast_today_high, + "sunrise": forecast.get("sunrise"), + "sunset": forecast.get("sunset"), + }, + "peak": result.get("peak") or {}, + "metar_status": result.get("metar_status") or {}, + } + try: + IntradayPathSnapshotRepository().append_snapshot(payload) + except Exception as exc: + logger.debug(f"intraday path snapshot archive skipped for {city}: {exc}") + + def _analyze( city: str, force_refresh: bool = False, @@ -2474,6 +2544,8 @@ def _analyze( "updated_at": datetime.now(timezone.utc).isoformat(), } result["intraday_meteorology"] = _build_intraday_meteorology(result) + if normalized_detail_mode == "full": + _archive_intraday_path_snapshot(city, result) if include_llm_commentary: result["dynamic_commentary"] = _maybe_enrich_dynamic_commentary_with_groq(