gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add ops dashboards for training data and model coverage

Browse Source
This commit is contained in:
2569718930@qq.com
2026-04-03 00:57:19 +08:00
parent 37cd8b8166
commit 781c247952
24 changed files with 3634 additions and 907 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/analysis/deb_algorithm.py
+224 -2
View File
@@ -1,13 +1,17 @@
import os
import json
from datetime import datetime, timedelta
from typing import Optional
import requests
from src.analysis.settlement_rounding import apply_city_settlement
from src.data_collection.wunderground_sources import fetch_wunderground_historical_high
from loguru import logger
from src.database.runtime_state import (
DailyRecordRepository,
STATE_STORAGE_DUAL,
STATE_STORAGE_SQLITE,
TrainingFeatureRecordRepository,
TruthRecordRepository,
get_state_storage_mode,
)
@@ -44,6 +48,9 @@ else:
_history_cache = {}
_history_mtime = 0
_daily_record_repo = DailyRecordRepository()
_training_feature_repo = TrainingFeatureRecordRepository()
_truth_record_repo = TruthRecordRepository()
_TRUTH_VERSION = "v1"
def _sf(value):
@@ -168,6 +175,80 @@ def _resolve_city_history_context(city_name: str):
return city_key, city_meta
def _truth_meta_for_city(city_meta: dict) -> dict:
if not isinstance(city_meta, dict):
city_meta = {}
return {
"settlement_source": str(city_meta.get("settlement_source") or "metar").strip().lower(),
"settlement_station_code": str(city_meta.get("settlement_station_code") or city_meta.get("icao") or "").strip().upper() or None,
"settlement_station_label": str(
city_meta.get("settlement_station_label")
or city_meta.get("airport_name")
or city_meta.get("name")
or ""
).strip()
or None,
}
def _persist_truth_record(
city_name: str,
date_str: str,
actual_high: float,
*,
city_meta: Optional[dict] = None,
updated_by: str,
reason: str,
source_payload: Optional[dict] = None,
is_final: bool = True,
) -> None:
city_key, resolved_meta = _resolve_city_history_context(city_name)
meta = city_meta if isinstance(city_meta, dict) else resolved_meta
if not city_key or not isinstance(meta, dict):
return
truth_meta = _truth_meta_for_city(meta)
_truth_record_repo.upsert_truth(
city=city_key,
target_date=date_str,
actual_high=float(actual_high),
settlement_source=truth_meta["settlement_source"],
settlement_station_code=truth_meta["settlement_station_code"],
settlement_station_label=truth_meta["settlement_station_label"],
truth_version=_TRUTH_VERSION,
updated_by=updated_by,
source_payload=source_payload,
is_final=is_final,
reason=reason,
)
def _persist_training_feature_record(
city_name: str,
date_str: str,
*,
forecasts: Optional[dict],
deb_prediction: Optional[float],
mu: Optional[float],
probability_features: Optional[dict],
probabilities: Optional[list],
shadow_probabilities: Optional[list],
probability_calibration: Optional[dict],
) -> None:
city_key, _ = _resolve_city_history_context(city_name)
if not city_key:
return
payload = {
"forecasts": forecasts or {},
"deb_prediction": deb_prediction,
"mu": mu,
"probability_features": probability_features or {},
"prob_snapshot": probabilities or [],
"shadow_prob_snapshot": shadow_probabilities or [],
"probability_calibration": probability_calibration or {},
}
_training_feature_repo.upsert_record(city_key, date_str, payload)
def _parse_hko_ryes_max_temp(payload):
if not isinstance(payload, dict):
return None
@@ -275,6 +356,15 @@ def _reconcile_recent_metar_actual_highs(city_name: str, lookback_days: int = 7)
if t_c is None:
continue
corrected = round(t_c * 9 / 5 + 32, 1) if use_fahrenheit else round(t_c, 1)
_persist_truth_record(
city_key,
d,
corrected,
city_meta=city_meta,
updated_by="backfill:metar_history",
reason="reconcile_recent_actual_highs",
source_payload={"icao": icao, "actual_high": corrected, "source": "metar"},
)
rec = city_data.get(d) or {}
old = rec.get("actual_high")
try:
@@ -367,6 +457,19 @@ def _reconcile_recent_hko_actual_highs(city_name: str, lookback_days: int = 14):
if use_fahrenheit
else round(max_temp_c, 1)
)
_persist_truth_record(
city_key,
date_str,
corrected,
city_meta=city_meta,
updated_by="backfill:hko_history",
reason="reconcile_recent_actual_highs",
source_payload={
"station_code": station_code,
"actual_high": corrected,
"source": "hko",
},
)
rec = city_data.get(date_str) or {}
old = rec.get("actual_high")
try:
@@ -486,6 +589,19 @@ def _reconcile_recent_noaa_actual_highs(city_name: str, lookback_days: int = 14)
if use_fahrenheit
else int(corrected)
)
_persist_truth_record(
city_key,
date_key,
next_value,
city_meta=city_meta,
updated_by="backfill:noaa_history",
reason="reconcile_recent_actual_highs",
source_payload={
"station_code": station_code,
"actual_high": next_value,
"source": "noaa",
},
)
rec = city_data.get(date_key) or {}
old = rec.get("actual_high")
try:
@@ -513,6 +629,79 @@ def _reconcile_recent_noaa_actual_highs(city_name: str, lookback_days: int = 14)
return {"ok": False, "reason": str(e), "updated": 0}
def _reconcile_recent_wunderground_actual_highs(city_name: str, lookback_days: int = 14):
try:
city_key, city_meta = _resolve_city_history_context(city_name)
if not city_key or not isinstance(city_meta, dict):
return {"ok": False, "reason": "unknown_city", "updated": 0}
settlement_url = str(city_meta.get("settlement_url") or "").strip()
if not settlement_url:
return {"ok": False, "reason": "missing_settlement_url", "updated": 0}
tz_offset = int(city_meta.get("tz_offset") or 0)
history_file = _get_history_file_path()
data = load_history(history_file)
city_data = data.get(city_key) or {}
if not isinstance(city_data, dict) or not city_data:
return {"ok": True, "reason": "no_city_history", "updated": 0}
local_now = datetime.utcnow() + timedelta(seconds=tz_offset)
local_today = local_now.strftime("%Y-%m-%d")
cutoff = (local_now - timedelta(days=max(lookback_days, 1) + 1)).strftime(
"%Y-%m-%d"
)
target_dates = sorted(
d for d in city_data.keys() if isinstance(d, str) and cutoff <= d < local_today
)
if not target_dates:
return {"ok": True, "reason": "no_target_dates", "updated": 0}
updated = 0
scanned_dates = 0
for date_str in target_dates:
result = fetch_wunderground_historical_high(city_key, date_str, url=settlement_url)
if not result.get("ok"):
continue
scanned_dates += 1
corrected = _sf(result.get("actual_high"))
if corrected is None:
continue
rec = city_data.get(date_str) or {}
old = rec.get("actual_high")
try:
old_val = float(old) if old is not None else None
except Exception:
old_val = None
if old_val is None or abs(old_val - corrected) >= 0.1:
rec["actual_high"] = corrected
city_data[date_str] = rec
updated += 1
_persist_truth_record(
city_key,
date_str,
corrected,
city_meta=city_meta,
updated_by="backfill:wunderground_history",
reason="reconcile_recent_actual_highs",
source_payload=result,
)
if updated > 0:
data[city_key] = city_data
save_history(history_file, data)
return {
"ok": True,
"updated": updated,
"scanned_dates": scanned_dates,
"station_code": city_meta.get("settlement_station_code"),
"source": "wunderground",
}
except Exception as e:
return {"ok": False, "reason": str(e), "updated": 0}
def reconcile_recent_actual_highs(city_name: str, lookback_days: int = 7):
"""
Reconcile recent `actual_high` values using the city's official settlement source.
@@ -526,6 +715,8 @@ def reconcile_recent_actual_highs(city_name: str, lookback_days: int = 7):
return _reconcile_recent_hko_actual_highs(city_key, lookback_days=lookback_days)
if settlement_source == "noaa":
return _reconcile_recent_noaa_actual_highs(city_key, lookback_days=lookback_days)
if settlement_source == "wunderground":
return _reconcile_recent_wunderground_actual_highs(city_key, lookback_days=lookback_days)
return _reconcile_recent_metar_actual_highs(city_key, lookback_days=lookback_days)
@@ -541,14 +732,14 @@ def bootstrap_recent_daily_history_if_missing(city_name: str, lookback_days: int
return {"ok": False, "reason": "unknown_city", "seeded": 0, "updated": 0}
settlement_source = str(city_meta.get("settlement_source") or "metar").strip().lower()
if settlement_source not in {"metar", "hko", "noaa"}:
if settlement_source not in {"metar", "hko", "noaa", "wunderground"}:
return {"ok": True, "reason": "unsupported_settlement_source", "seeded": 0, "updated": 0}
icao = str(city_meta.get("icao") or "").strip().upper()
station_code = str(city_meta.get("settlement_station_code") or "").strip().upper()
if settlement_source == "metar" and not icao:
return {"ok": False, "reason": "missing_icao", "seeded": 0, "updated": 0}
if settlement_source in {"hko", "noaa"} and not station_code:
if settlement_source in {"hko", "noaa", "wunderground"} and not station_code:
return {"ok": False, "reason": "missing_station_code", "seeded": 0, "updated": 0}
tz_offset = int(city_meta.get("tz_offset") or 0)
@@ -726,6 +917,37 @@ def update_daily_record(
if compact_calibration is not None:
existing["probability_calibration"] = compact_calibration
if actual_high is not None:
try:
_persist_truth_record(
city_name,
date_str,
float(actual_high),
updated_by="runtime:update_daily_record",
reason="update_daily_record",
source_payload={
"actual_high": actual_high,
"deb_prediction": deb_prediction,
"mu": next_mu,
},
)
except Exception as e:
logger.error(f"Error persisting truth record city={city_name} date={date_str}: {e}")
try:
_persist_training_feature_record(
city_name,
date_str,
forecasts=merged_forecasts,
deb_prediction=existing.get("deb_prediction"),
mu=existing.get("mu"),
probability_features=existing.get("probability_features"),
probabilities=existing.get("prob_snapshot"),
shadow_probabilities=existing.get("shadow_prob_snapshot"),
probability_calibration=existing.get("probability_calibration"),
)
except Exception as e:
logger.error(f"Error persisting training feature record city={city_name} date={date_str}: {e}")
# 自动清理：只保留最近 14 天的记录（DEB 只用 7 天，14 天留足余量）
cutoff = (datetime.now() - timedelta(days=14)).strftime("%Y-%m-%d")
for city in list(data.keys()):
src/analysis/probability_snapshot_archive.py
+32
View File
@@ -9,6 +9,7 @@ from src.database.runtime_state import (
ProbabilitySnapshotRepository,
STATE_STORAGE_DUAL,
STATE_STORAGE_SQLITE,
TrainingFeatureRecordRepository,
get_state_storage_mode,
)
@@ -17,6 +18,7 @@ MU_THRESHOLD = 0.2
SIGMA_THRESHOLD = 0.15
MAX_SO_FAR_THRESHOLD = 0.2
_snapshot_repo = ProbabilitySnapshotRepository()
_training_feature_repo = TrainingFeatureRecordRepository()
def _sf(value: Any) -> Optional[float]:
@@ -260,6 +262,36 @@ def append_probability_snapshot(
mode = get_state_storage_mode()
if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
_snapshot_repo.append_snapshot(payload)
_training_feature_repo.upsert_record(
city_key,
local_date,
{
"forecasts": payload.get("multi_model") or {},
"deb_prediction": payload.get("deb_prediction"),
"mu": payload.get("raw_mu"),
"probability_features": {
"raw_mu": payload.get("raw_mu"),
"raw_sigma": payload.get("raw_sigma"),
"deb_prediction": payload.get("deb_prediction"),
"ens_median": (payload.get("ensemble") or {}).get("median"),
"ensemble_spread": None,
"max_so_far": payload.get("max_so_far"),
"peak_status": payload.get("peak_status"),
},
"prob_snapshot": payload.get("prob_snapshot") or [],
"shadow_prob_snapshot": payload.get("shadow_prob_snapshot") or [],
"probability_calibration": {
"engine": payload.get("probability_engine"),
"mode": payload.get("probability_mode"),
"calibration_version": payload.get("calibration_version"),
"calibration_source": payload.get("calibration_source"),
"calibrated_mu": payload.get("calibrated_mu"),
"calibrated_sigma": payload.get("calibrated_sigma"),
},
"observation": payload.get("observation") or {},
"snapshot_timestamp": payload.get("timestamp"),
},
)
if mode != STATE_STORAGE_SQLITE:
with open(path, "a", encoding="utf-8") as fh:
src/data_collection/wunderground_sources.py
+91
View File
@@ -5,8 +5,10 @@ import re
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional
import requests
from loguru import logger
from src.data_collection.city_registry import CITY_REGISTRY
from src.analysis.settlement_rounding import apply_city_settlement
class WundergroundSourceMixin:
@@ -502,3 +504,92 @@ class WundergroundSourceMixin:
}
self._set_settlement_cache(cache_key, payload)
return payload
def _normalize_wu_history_date_url(url: str, target_date: str) -> str:
normalized = str(url or "").strip().rstrip("/")
normalized = re.sub(r"/date/\d{4}-\d{2}-\d{2}$", "", normalized, flags=re.IGNORECASE)
return f"{normalized}/date/{target_date}"
def fetch_wunderground_historical_high(
city: str,
target_date: str,
*,
url: Optional[str] = None,
timeout: int = 15,
session: Optional[requests.Session] = None,
) -> Dict[str, Any]:
city_key = str(city or "").strip().lower()
city_meta = CITY_REGISTRY.get(city_key) or {}
history_url = _normalize_wu_history_date_url(
url or str(city_meta.get("settlement_url") or "").strip(),
target_date,
)
if not history_url:
return {"ok": False, "reason": "missing_history_url", "city": city_key, "date": target_date}
requester = session or requests.Session()
try:
response = requester.get(
history_url,
headers={
"User-Agent": "Mozilla/5.0",
"Referer": history_url,
},
timeout=timeout,
)
response.raise_for_status()
html = str(response.text or "")
except Exception as exc:
logger.warning(f"Wunderground history fetch failed city={city_key} date={target_date}: {exc}")
return {
"ok": False,
"reason": "fetch_failed",
"city": city_key,
"date": target_date,
"history_url": history_url,
"error": str(exc),
}
app_state = WundergroundSourceMixin._wu_extract_app_state(html)
if not isinstance(app_state, dict):
return {
"ok": False,
"reason": "missing_app_state",
"city": city_key,
"date": target_date,
"history_url": history_url,
}
utc_offset_seconds = int(city_meta.get("tz_offset") or 0)
obs = WundergroundSourceMixin._wu_extract_history_observations(
app_state,
utc_offset_seconds=utc_offset_seconds,
)
if not obs:
return {
"ok": False,
"reason": "missing_observations",
"city": city_key,
"date": target_date,
"history_url": history_url,
}
raw_max_temp_c = max(float(point.get("temp")) for point in obs if point.get("temp") is not None)
settled_actual_high = apply_city_settlement(city_key, raw_max_temp_c)
station_code = str(city_meta.get("settlement_station_code") or "").strip().upper() or None
station_label = str(city_meta.get("settlement_station_label") or "").strip() or None
return {
"ok": True,
"city": city_key,
"date": target_date,
"history_url": history_url,
"raw_max_temp_c": round(raw_max_temp_c, 1),
"actual_high": float(settled_actual_high),
"settlement_source": "wunderground",
"settlement_station_code": station_code,
"settlement_station_label": station_label,
"observation_count": len(obs),
"observations": obs,
}
src/database/runtime_state.py
+361
View File
@@ -74,6 +74,48 @@ class RuntimeStateDB:
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS truth_records_store (
city TEXT NOT NULL,
target_date TEXT NOT NULL,
actual_high REAL NOT NULL,
settlement_source TEXT,
settlement_station_code TEXT,
settlement_station_label TEXT,
truth_version TEXT,
updated_by TEXT,
updated_at REAL NOT NULL,
source_payload_json TEXT,
is_final INTEGER NOT NULL DEFAULT 1,
PRIMARY KEY (city, target_date)
)
"""
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS truth_revisions_store (
id INTEGER PRIMARY KEY AUTOINCREMENT,
city TEXT NOT NULL,
target_date TEXT NOT NULL,
previous_actual_high REAL,
next_actual_high REAL NOT NULL,
previous_source TEXT,
next_source TEXT,
truth_version TEXT,
updated_by TEXT,
updated_at REAL NOT NULL,
reason TEXT,
payload_json TEXT
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_truth_records_city_date ON truth_records_store(city, target_date)"
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_truth_revisions_city_date ON truth_revisions_store(city, target_date, id DESC)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS telegram_alert_last_by_city (
@@ -117,6 +159,20 @@ class RuntimeStateDB:
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_probability_snapshot_city_date ON probability_training_snapshots_store(city, target_date, id DESC)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS training_feature_records_store (
city TEXT NOT NULL,
target_date TEXT NOT NULL,
updated_at REAL NOT NULL,
payload_json TEXT NOT NULL,
PRIMARY KEY (city, target_date)
)
"""
)
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_training_feature_records_city_date ON training_feature_records_store(city, target_date)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS open_meteo_cache_store (
@@ -238,6 +294,251 @@ class DailyRecordRepository:
return int(cur.rowcount or 0)
class TruthRecordRepository:
def __init__(self, db: Optional[RuntimeStateDB] = None):
self.db = db or RuntimeStateDB.instance()
def load_all(self) -> Dict[str, Dict[str, Dict[str, Any]]]:
out: Dict[str, Dict[str, Dict[str, Any]]] = {}
with self.db.connect() as conn:
rows = conn.execute(
"""
SELECT city, target_date, actual_high, settlement_source, settlement_station_code,
settlement_station_label, truth_version, updated_by, updated_at,
source_payload_json, is_final
FROM truth_records_store
ORDER BY city, target_date
"""
).fetchall()
for row in rows:
payload: Dict[str, Any] = {
"actual_high": float(row["actual_high"]),
"settlement_source": row["settlement_source"],
"settlement_station_code": row["settlement_station_code"],
"settlement_station_label": row["settlement_station_label"],
"truth_version": row["truth_version"],
"updated_by": row["updated_by"],
"truth_updated_at": float(row["updated_at"]),
"is_final": bool(row["is_final"]),
}
if row["source_payload_json"]:
try:
payload["source_payload"] = json.loads(row["source_payload_json"])
except Exception:
pass
out.setdefault(str(row["city"]), {})[str(row["target_date"])] = payload
return out
def get_record(self, city: str, target_date: str) -> Optional[Dict[str, Any]]:
with self.db.connect() as conn:
row = conn.execute(
"""
SELECT actual_high, settlement_source, settlement_station_code,
settlement_station_label, truth_version, updated_by, updated_at,
source_payload_json, is_final
FROM truth_records_store
WHERE city = ? AND target_date = ?
""",
(city, target_date),
).fetchone()
if not row:
return None
payload: Dict[str, Any] = {
"actual_high": float(row["actual_high"]),
"settlement_source": row["settlement_source"],
"settlement_station_code": row["settlement_station_code"],
"settlement_station_label": row["settlement_station_label"],
"truth_version": row["truth_version"],
"updated_by": row["updated_by"],
"truth_updated_at": float(row["updated_at"]),
"is_final": bool(row["is_final"]),
}
if row["source_payload_json"]:
try:
payload["source_payload"] = json.loads(row["source_payload_json"])
except Exception:
pass
return payload
def upsert_truth(
self,
*,
city: str,
target_date: str,
actual_high: float,
settlement_source: Optional[str],
settlement_station_code: Optional[str],
settlement_station_label: Optional[str],
truth_version: str,
updated_by: str,
source_payload: Optional[Dict[str, Any]] = None,
is_final: bool = True,
reason: Optional[str] = None,
) -> bool:
updated_at = time.time()
payload_json = (
json.dumps(source_payload, ensure_ascii=False) if source_payload is not None else None
)
with self.db.connect() as conn:
current = conn.execute(
"""
SELECT actual_high, settlement_source, source_payload_json
FROM truth_records_store
WHERE city = ? AND target_date = ?
""",
(city, target_date),
).fetchone()
changed = True
if current:
prev_actual = float(current["actual_high"])
prev_source = str(current["settlement_source"] or "")
next_source = str(settlement_source or "")
changed = (
abs(prev_actual - float(actual_high)) >= 0.0001
or prev_source != next_source
or str(current["source_payload_json"] or "") != str(payload_json or "")
)
if changed:
conn.execute(
"""
INSERT INTO truth_revisions_store (
city, target_date, previous_actual_high, next_actual_high,
previous_source, next_source, truth_version, updated_by,
updated_at, reason, payload_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
city,
target_date,
prev_actual,
float(actual_high),
prev_source or None,
next_source or None,
truth_version,
updated_by,
updated_at,
reason,
payload_json,
),
)
conn.execute(
"""
INSERT INTO truth_records_store (
city, target_date, actual_high, settlement_source,
settlement_station_code, settlement_station_label, truth_version,
updated_by, updated_at, source_payload_json, is_final
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(city, target_date) DO UPDATE SET
actual_high = excluded.actual_high,
settlement_source = excluded.settlement_source,
settlement_station_code = excluded.settlement_station_code,
settlement_station_label = excluded.settlement_station_label,
truth_version = excluded.truth_version,
updated_by = excluded.updated_by,
updated_at = excluded.updated_at,
source_payload_json = excluded.source_payload_json,
is_final = excluded.is_final
""",
(
city,
target_date,
float(actual_high),
settlement_source,
settlement_station_code,
settlement_station_label,
truth_version,
updated_by,
updated_at,
payload_json,
1 if is_final else 0,
),
)
conn.commit()
return changed
def replace_all(self, rows: Dict[str, Dict[str, Dict[str, Any]]]) -> int:
count = 0
with self.db.connect() as conn:
conn.execute("DELETE FROM truth_records_store")
conn.execute("DELETE FROM truth_revisions_store")
for city, city_rows in (rows or {}).items():
if not isinstance(city_rows, dict):
continue
for target_date, record in city_rows.items():
if not isinstance(record, dict):
continue
actual_high = record.get("actual_high")
if actual_high is None:
continue
payload_json = (
json.dumps(record.get("source_payload"), ensure_ascii=False)
if record.get("source_payload") is not None
else None
)
conn.execute(
"""
INSERT INTO truth_records_store (
city, target_date, actual_high, settlement_source,
settlement_station_code, settlement_station_label, truth_version,
updated_by, updated_at, source_payload_json, is_final
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
city,
target_date,
float(actual_high),
record.get("settlement_source"),
record.get("settlement_station_code"),
record.get("settlement_station_label"),
record.get("truth_version") or "v1",
record.get("updated_by") or "replace_all",
float(record.get("truth_updated_at") or time.time()),
payload_json,
1 if record.get("is_final", True) else 0,
),
)
count += 1
conn.commit()
return count
class TruthRevisionRepository:
def __init__(self, db: Optional[RuntimeStateDB] = None):
self.db = db or RuntimeStateDB.instance()
def load_revisions(self, city: str, target_date: str) -> List[Dict[str, Any]]:
with self.db.connect() as conn:
rows = conn.execute(
"""
SELECT previous_actual_high, next_actual_high, previous_source, next_source,
truth_version, updated_by, updated_at, reason, payload_json
FROM truth_revisions_store
WHERE city = ? AND target_date = ?
ORDER BY id ASC
""",
(city, target_date),
).fetchall()
out: List[Dict[str, Any]] = []
for row in rows:
entry: Dict[str, Any] = {
"previous_actual_high": row["previous_actual_high"],
"next_actual_high": row["next_actual_high"],
"previous_source": row["previous_source"],
"next_source": row["next_source"],
"truth_version": row["truth_version"],
"updated_by": row["updated_by"],
"updated_at": float(row["updated_at"]),
"reason": row["reason"],
}
if row["payload_json"]:
try:
entry["payload"] = json.loads(row["payload_json"])
except Exception:
pass
out.append(entry)
return out
class TelegramAlertStateRepository:
def __init__(self, db: Optional[RuntimeStateDB] = None):
self.db = db or RuntimeStateDB.instance()
@@ -429,6 +730,66 @@ class ProbabilitySnapshotRepository:
return count
class TrainingFeatureRecordRepository:
def __init__(self, db: Optional[RuntimeStateDB] = None):
self.db = db or RuntimeStateDB.instance()
def upsert_record(self, city: str, target_date: str, payload: Dict[str, Any]) -> None:
with self.db.connect() as conn:
conn.execute(
"""
INSERT INTO training_feature_records_store (
city, target_date, updated_at, payload_json
) VALUES (?, ?, ?, ?)
ON CONFLICT(city, target_date) DO UPDATE SET
updated_at = excluded.updated_at,
payload_json = excluded.payload_json
""",
(
city,
target_date,
time.time(),
json.dumps(payload, ensure_ascii=False),
),
)
conn.commit()
def load_all(self) -> Dict[str, Dict[str, Dict[str, Any]]]:
out: Dict[str, Dict[str, Dict[str, Any]]] = {}
with self.db.connect() as conn:
rows = conn.execute(
"""
SELECT city, target_date, payload_json
FROM training_feature_records_store
ORDER BY city, target_date
"""
).fetchall()
for row in rows:
try:
payload = json.loads(row["payload_json"])
except Exception:
continue
out.setdefault(str(row["city"]), {})[str(row["target_date"])] = payload
return out
def get_record(self, city: str, target_date: str) -> Optional[Dict[str, Any]]:
with self.db.connect() as conn:
row = conn.execute(
"""
SELECT payload_json
FROM training_feature_records_store
WHERE city = ? AND target_date = ?
""",
(city, target_date),
).fetchone()
if not row:
return None
try:
return json.loads(row["payload_json"])
except Exception:
return None
class OpenMeteoCacheRepository:
def __init__(self, db: Optional[RuntimeStateDB] = None):
self.db = db or RuntimeStateDB.instance()
src/models/lgbm_features.py
+84 -6
View File
@@ -13,6 +13,8 @@ from src.database.runtime_state import (
ProbabilitySnapshotRepository,
STATE_STORAGE_FILE,
STATE_STORAGE_SQLITE,
TrainingFeatureRecordRepository,
TruthRecordRepository,
get_state_storage_mode,
)
@@ -288,23 +290,93 @@ def build_training_samples(
snapshot_index: Optional[Dict[Tuple[str, str], Dict[str, Any]]] = None,
) -> List[Dict[str, Any]]:
if isinstance(history_data, dict):
data = history_data
runtime_history = history_data
elif get_state_storage_mode() == STATE_STORAGE_SQLITE:
data = DailyRecordRepository().load_all()
runtime_history = DailyRecordRepository().load_all()
else:
data = load_history(_history_file_path())
runtime_history = load_history(_history_file_path())
if get_state_storage_mode() == STATE_STORAGE_SQLITE:
truth_history = TruthRecordRepository().load_all()
training_feature_history = TrainingFeatureRecordRepository().load_all()
else:
truth_history = runtime_history
training_feature_history = {}
snapshots = snapshot_index if isinstance(snapshot_index, dict) else load_snapshot_index()
samples: List[Dict[str, Any]] = []
excluded_keys: set[tuple[str, str]] = set()
for city_name, city_records in (data or {}).items():
for (city_name, date_str), snapshot in (snapshots or {}).items():
if not isinstance(snapshot, dict):
continue
truth_row = ((truth_history.get(city_name) or {}).get(str(date_str)) or {})
target = _sf(truth_row.get("actual_high"))
if target is None:
runtime_record = ((runtime_history.get(city_name) or {}).get(str(date_str)) or {})
target = _sf(runtime_record.get("actual_high"))
if target is None:
continue
observation = snapshot.get("observation") if isinstance(snapshot.get("observation"), dict) else {}
current_forecasts = snapshot.get("multi_model") if isinstance(snapshot.get("multi_model"), dict) else {}
local_hour = _sf(observation.get("local_hour"))
if local_hour is None:
timestamp = _parse_timestamp(snapshot.get("timestamp"))
local_hour = float(timestamp.hour) if timestamp is not None else 12.0
feature_map, meta = build_runtime_feature_map(
city_name=city_name,
current_forecasts=current_forecasts,
deb_prediction=_sf(snapshot.get("deb_prediction")) or _sf(snapshot.get("raw_mu")),
current_temp=_sf(observation.get("current_temp")),
max_so_far=_sf(snapshot.get("max_so_far")),
humidity=_sf(observation.get("humidity")),
wind_speed_kt=_sf(observation.get("wind_speed_kt")),
visibility_mi=_sf(observation.get("visibility_mi")),
local_hour=int(local_hour),
local_date=str(date_str),
peak_status=str(snapshot.get("peak_status") or "before"),
history_data=truth_history,
)
if not feature_map:
continue
samples.append(
{
"city": _normalized_city_key(city_name),
"date": str(date_str),
"target": float(target),
"features": feature_map,
"vector": _features_to_vector(feature_map),
"history_count": int(meta.get("history_count") or 0),
"deb_prediction": _sf(snapshot.get("deb_prediction")) or _sf(snapshot.get("raw_mu")),
"forecasts": {
key: _sf(value)
for key, value in current_forecasts.items()
if _sf(value) is not None
},
"sample_source": "snapshot",
"settlement_source": truth_row.get("settlement_source"),
"settlement_station_code": truth_row.get("settlement_station_code"),
"truth_version": truth_row.get("truth_version"),
"truth_updated_by": truth_row.get("updated_by"),
"truth_updated_at": truth_row.get("truth_updated_at"),
}
)
excluded_keys.add((_normalized_city_key(city_name), str(date_str)))
training_source = training_feature_history or runtime_history or {}
for city_name, city_records in training_source.items():
if not isinstance(city_records, dict):
continue
ordered_dates = sorted(city_records.keys())
for date_str in ordered_dates:
normalized_city = _normalized_city_key(city_name)
if (normalized_city, str(date_str)) in excluded_keys:
continue
record = city_records.get(date_str)
if not isinstance(record, dict):
continue
target = _sf(record.get("actual_high"))
truth_row = ((truth_history.get(normalized_city) or {}).get(str(date_str)) or {})
target = _sf(truth_row.get("actual_high"))
if target is None:
target = _sf(((runtime_history.get(normalized_city) or {}).get(str(date_str)) or {}).get("actual_high"))
forecasts = record.get("forecasts") if isinstance(record.get("forecasts"), dict) else {}
if target is None or not forecasts:
continue
@@ -321,7 +393,7 @@ def build_training_samples(
local_hour=12,
local_date=str(date_str),
peak_status="before",
history_data=data,
history_data=truth_history,
)
if not feature_map:
continue
@@ -361,6 +433,12 @@ def build_training_samples(
for key, value in forecasts.items()
if _sf(value) is not None
},
"sample_source": "daily_record",
"settlement_source": truth_row.get("settlement_source"),
"settlement_station_code": truth_row.get("settlement_station_code"),
"truth_version": truth_row.get("truth_version"),
"truth_updated_by": truth_row.get("updated_by"),
"truth_updated_at": truth_row.get("truth_updated_at"),
}
)
samples.sort(key=lambda row: (row["date"], row["city"]))