From b3ea8dcfa79966c5ff6c06ad3baa887f3f68e0c0 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Sun, 10 May 2026 16:54:48 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E9=93=BE=E8=B7=AF=20P1=20?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9AETag=20=E7=BC=93=E5=AD=98=20+=20?= =?UTF-8?q?=E6=A0=A1=E5=87=86=E6=BC=82=E7=A7=BB=E6=A3=80=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1-5 ETag 支持： - 后端新增 _etag_middleware：GET /api/* 自动返回 ETag (MD5) - 支持 If-None-Match 请求头，匹配时返回 304 + 30s Cache-Control - 前端 cache: no-store → default，浏览器自动处理 ETag/304 节省带宽 P1-6 校准漂移检测： - probability_calibration.py 新增 check_calibration_drift() - 对比最近 200 条 daily_records 的 CRPS 与校准基线 - 漂移 >15% 时返回 warning 提示重新训练 - 集成到 /api/system/status 的 probability.drift 字段 Tested: python -m ruff check ., npx tsc --noEmit --- frontend/lib/dashboard-client.ts | 6 +-- src/analysis/probability_calibration.py | 72 +++++++++++++++++++++++++ web/analysis_service.py | 5 +- web/core.py | 58 ++++++++++++++++++++ 4 files changed, 135 insertions(+), 6 deletions(-) diff --git a/frontend/lib/dashboard-client.ts b/frontend/lib/dashboard-client.ts index 3385e13e..ff942166 100644 --- a/frontend/lib/dashboard-client.ts +++ b/frontend/lib/dashboard-client.ts @@ -68,7 +68,7 @@ async function fetchJson(url: string, options?: { timeoutMs?: number }): Prom try { response = await fetchBackendApi(url, { headers, - cache: "no-store", + cache: "default", signal: controller?.signal, }); } catch (error) { @@ -238,7 +238,7 @@ export const dashboardClient = { void fetch(`/api/system/priority-warm?${params.toString()}`, { method: "POST", headers: { Accept: "application/json" }, - cache: "no-store", + cache: "default", keepalive: true, }).catch(() => {}); }, @@ -418,7 +418,7 @@ export const dashboardClient = { }).then((headers) => fetchBackendApi("/api/scan/terminal/ai", { method: "POST", headers, - cache: "no-store", + cache: "default", body: JSON.stringify({ filters: payload.filters, snapshot_id: snapshotId || null, diff --git a/src/analysis/probability_calibration.py b/src/analysis/probability_calibration.py index 7bc8480c..d1897f86 100644 --- a/src/analysis/probability_calibration.py +++ b/src/analysis/probability_calibration.py @@ -760,3 +760,75 @@ def default_calibration_payload( "reason": reason, }, } + + +def check_calibration_drift(records: list[dict[str, Any]], calibration_path: str | None = None) -> dict[str, Any]: + """Compare recent CRPS against calibration baseline to detect drift. + + Returns a dict with keys: drifted (bool), current_crps (float), + baseline_crps (float), delta_pct (float), sample_count (int), warning (str|None). + A positive delta_pct means the model is performing worse than baseline. + """ + import json + import os + + if len(records) < 5: + return {"drifted": False, "sample_count": len(records), "warning": "Insufficient samples"} + + path = calibration_path or DEFAULT_CALIBRATION_FILE + baseline_crps = None + try: + with open(path, "r", encoding="utf-8") as fh: + json.load(fh) # calibration file + eval_path = os.path.join( + os.path.dirname(path) if os.path.dirname(path) else os.path.join(os.path.dirname(DEFAULT_CALIBRATION_FILE)), + "evaluation_report.json", + ) + with open(eval_path, "r", encoding="utf-8") as fh: + report = json.load(fh) + baseline_crps = float(report.get("metrics", {}).get("selected_mean_crps") or 0) or None + except Exception: + pass + + current_crps_values: list[float] = [] + for r in records: + actual = float(r.get("actual_high") or r.get("observed") or 0) + mu = float(r.get("deb_prediction") or r.get("mu") or 0) + sigma = float(r.get("sigma") or r.get("ensemble_std") or 2.0) + if actual and mu and sigma > 0: + current_crps_values.append(_gaussian_crps(actual, mu, sigma)) + + if not current_crps_values: + return {"drifted": False, "sample_count": 0, "warning": "No valid records for CRPS"} + + current_crps = round(sum(current_crps_values) / len(current_crps_values), 6) + + if baseline_crps is None or baseline_crps <= 0: + return { + "drifted": False, + "current_crps": current_crps, + "baseline_crps": baseline_crps, + "sample_count": len(records), + "warning": "No baseline available", + } + + delta_pct = round((current_crps - baseline_crps) / baseline_crps * 100, 1) + DRIFT_THRESHOLD_PCT = 15.0 # warn if CRPS degraded by >15% + + if delta_pct > DRIFT_THRESHOLD_PCT: + return { + "drifted": True, + "current_crps": current_crps, + "baseline_crps": baseline_crps, + "delta_pct": delta_pct, + "sample_count": len(records), + "warning": f"CRPS degraded {delta_pct}% vs baseline; consider re-running fit_calibration()", + } + + return { + "drifted": False, + "current_crps": current_crps, + "baseline_crps": baseline_crps, + "delta_pct": delta_pct, + "sample_count": len(records), + } diff --git a/web/analysis_service.py b/web/analysis_service.py index ffdb3643..b228e726 100644 --- a/web/analysis_service.py +++ b/web/analysis_service.py @@ -2069,9 +2069,8 @@ def _analyze( local_date=local_date_str, peak_status=peak_status, ) - if lgbm_val is not None: - # LGBM is kept as an independent reference, not fed back into DEB - # to avoid circular dependency (DEB → LGBM training → DEB) + # LGBM is kept as an independent reference (lgbm.prediction), + # not fed back into DEB to avoid circular dependency deviation_monitor = _build_deviation_monitor( current_temp=cur_temp, diff --git a/web/core.py b/web/core.py index bec62ea5..b5067a07 100644 --- a/web/core.py +++ b/web/core.py @@ -30,6 +30,7 @@ from src.utils.metrics import ( ) from src.analysis.probability_calibration import ( DEFAULT_CALIBRATION_FILE, + check_calibration_drift, resolve_probability_engine_mode, ) from src.analysis.probability_rollout import build_rollout_report @@ -132,6 +133,40 @@ async def _metrics_middleware(request: Request, call_next): status=status_code, ) return response + + +@app.middleware("http") +async def _etag_middleware(request: Request, call_next): + """Add ETag to GET /api/* responses; return 304 on If-None-Match hit.""" + response = await call_next(request) + + if request.method != "GET" or response.status_code != 200: + return response + + path = request.url.path + if not path.startswith("/api/") or path.endswith("/stream"): + return response + + body = getattr(response, "body", None) or b"" + if not body: + return response + + try: + import hashlib + etag = hashlib.md5(body).hexdigest() + except Exception: + return response + + etag_value = f'"{etag}"' + if_none_match = request.headers.get("If-None-Match", "") + if if_none_match == etag_value: + from fastapi.responses import Response + return Response(status_code=304, headers={"ETag": etag_value}) + + response.headers["ETag"] = etag_value + response.headers["Cache-Control"] = "private, max-age=30" + return response + _PROBABILITY_SHADOW_REPORT = os.path.join( _PROJECT_ROOT, "artifacts", @@ -453,11 +488,34 @@ def _probability_summary() -> Dict[str, Any]: _PROBABILITY_EVALUATION_REPORT, _PROBABILITY_SHADOW_REPORT, ) + drift = {"drifted": False, "sample_count": 0, "warning": "not checked"} + try: + import sqlite3 + conn = sqlite3.connect(_account_db.db_path) + rows = conn.execute( + "SELECT city, target_date, actual_high, deb_prediction, extra_json FROM daily_records_store ORDER BY target_date DESC LIMIT 200" + ).fetchall() + conn.close() + records: list[dict[str, Any]] = [] + for row in rows: + rec = {"city": row[0], "target_date": row[1], "actual_high": row[2], "deb_prediction": row[3]} + try: + extra = json.loads(row[4] or "{}") if row[4] else {} + rec["mu"] = extra.get("mu") + rec["sigma"] = extra.get("sigma") + except Exception: + pass + records.append(rec) + if records: + drift = check_calibration_drift(records) + except Exception: + pass return { "engine_mode": resolve_probability_engine_mode(), "calibration_file": os.getenv("POLYWEATHER_PROBABILITY_CALIBRATION_FILE") or DEFAULT_CALIBRATION_FILE, "rollout": rollout, + "drift": drift, }