数据链路 P1 修复:ETag 缓存 + 校准漂移检测
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
This commit is contained in:
@@ -68,7 +68,7 @@ async function fetchJson<T>(url: string, options?: { timeoutMs?: number }): Prom
|
|||||||
try {
|
try {
|
||||||
response = await fetchBackendApi(url, {
|
response = await fetchBackendApi(url, {
|
||||||
headers,
|
headers,
|
||||||
cache: "no-store",
|
cache: "default",
|
||||||
signal: controller?.signal,
|
signal: controller?.signal,
|
||||||
});
|
});
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -238,7 +238,7 @@ export const dashboardClient = {
|
|||||||
void fetch(`/api/system/priority-warm?${params.toString()}`, {
|
void fetch(`/api/system/priority-warm?${params.toString()}`, {
|
||||||
method: "POST",
|
method: "POST",
|
||||||
headers: { Accept: "application/json" },
|
headers: { Accept: "application/json" },
|
||||||
cache: "no-store",
|
cache: "default",
|
||||||
keepalive: true,
|
keepalive: true,
|
||||||
}).catch(() => {});
|
}).catch(() => {});
|
||||||
},
|
},
|
||||||
@@ -418,7 +418,7 @@ export const dashboardClient = {
|
|||||||
}).then((headers) => fetchBackendApi("/api/scan/terminal/ai", {
|
}).then((headers) => fetchBackendApi("/api/scan/terminal/ai", {
|
||||||
method: "POST",
|
method: "POST",
|
||||||
headers,
|
headers,
|
||||||
cache: "no-store",
|
cache: "default",
|
||||||
body: JSON.stringify({
|
body: JSON.stringify({
|
||||||
filters: payload.filters,
|
filters: payload.filters,
|
||||||
snapshot_id: snapshotId || null,
|
snapshot_id: snapshotId || null,
|
||||||
|
|||||||
@@ -760,3 +760,75 @@ def default_calibration_payload(
|
|||||||
"reason": reason,
|
"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),
|
||||||
|
}
|
||||||
|
|||||||
@@ -2069,9 +2069,8 @@ def _analyze(
|
|||||||
local_date=local_date_str,
|
local_date=local_date_str,
|
||||||
peak_status=peak_status,
|
peak_status=peak_status,
|
||||||
)
|
)
|
||||||
if lgbm_val is not None:
|
# LGBM is kept as an independent reference (lgbm.prediction),
|
||||||
# LGBM is kept as an independent reference, not fed back into DEB
|
# not fed back into DEB to avoid circular dependency
|
||||||
# to avoid circular dependency (DEB → LGBM training → DEB)
|
|
||||||
|
|
||||||
deviation_monitor = _build_deviation_monitor(
|
deviation_monitor = _build_deviation_monitor(
|
||||||
current_temp=cur_temp,
|
current_temp=cur_temp,
|
||||||
|
|||||||
+58
@@ -30,6 +30,7 @@ from src.utils.metrics import (
|
|||||||
)
|
)
|
||||||
from src.analysis.probability_calibration import (
|
from src.analysis.probability_calibration import (
|
||||||
DEFAULT_CALIBRATION_FILE,
|
DEFAULT_CALIBRATION_FILE,
|
||||||
|
check_calibration_drift,
|
||||||
resolve_probability_engine_mode,
|
resolve_probability_engine_mode,
|
||||||
)
|
)
|
||||||
from src.analysis.probability_rollout import build_rollout_report
|
from src.analysis.probability_rollout import build_rollout_report
|
||||||
@@ -132,6 +133,40 @@ async def _metrics_middleware(request: Request, call_next):
|
|||||||
status=status_code,
|
status=status_code,
|
||||||
)
|
)
|
||||||
return response
|
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(
|
_PROBABILITY_SHADOW_REPORT = os.path.join(
|
||||||
_PROJECT_ROOT,
|
_PROJECT_ROOT,
|
||||||
"artifacts",
|
"artifacts",
|
||||||
@@ -453,11 +488,34 @@ def _probability_summary() -> Dict[str, Any]:
|
|||||||
_PROBABILITY_EVALUATION_REPORT,
|
_PROBABILITY_EVALUATION_REPORT,
|
||||||
_PROBABILITY_SHADOW_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 {
|
return {
|
||||||
"engine_mode": resolve_probability_engine_mode(),
|
"engine_mode": resolve_probability_engine_mode(),
|
||||||
"calibration_file": os.getenv("POLYWEATHER_PROBABILITY_CALIBRATION_FILE")
|
"calibration_file": os.getenv("POLYWEATHER_PROBABILITY_CALIBRATION_FILE")
|
||||||
or DEFAULT_CALIBRATION_FILE,
|
or DEFAULT_CALIBRATION_FILE,
|
||||||
"rollout": rollout,
|
"rollout": rollout,
|
||||||
|
"drift": drift,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user