Unify runtime state in SQLite and add rollout observability

src/analysis/deb_algorithm.py

@@ -3,6 +3,13 @@ import json
 from datetime import datetime, timedelta
 import requests
 from src.analysis.settlement_rounding import apply_city_settlement
+from loguru import logger
+from src.database.runtime_state import (
+    DailyRecordRepository,
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+    get_state_storage_mode,
+)
 
 # Cross-platform file locking
 import sys
@@ -36,6 +43,7 @@ else:
 # Simple memory cache to avoid blasting the disk if queried 10 times a minute
 _history_cache = {}
 _history_mtime = 0
+_daily_record_repo = DailyRecordRepository()
 
 
 def _sf(value):
@@ -54,7 +62,24 @@ def _is_excluded_model_name(model_name: str) -> bool:
 
 def load_history(filepath):
     global _history_cache, _history_mtime
+    mode = get_state_storage_mode()
+
+    if mode == STATE_STORAGE_SQLITE:
+        try:
+            data = _daily_record_repo.load_all()
+            _history_cache = data
+            return data
+        except Exception as e:
+            logger.error(f"Error loading daily records from sqlite, fallback to file: {e}")
+
     if not os.path.exists(filepath):
+        if mode == STATE_STORAGE_DUAL:
+            try:
+                data = _daily_record_repo.load_all()
+                _history_cache = data
+                return data
+            except Exception:
+                return {}
         return {}
 
     try:
@@ -80,6 +105,18 @@ def load_history(filepath):
 def save_history(filepath, data):
     global _history_cache, _history_mtime
     _history_cache = data
+    mode = get_state_storage_mode()
+
+    if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+        try:
+            _daily_record_repo.replace_all(data)
+        except Exception as e:
+            logger.error(f"Error saving daily records to sqlite: {e}")
+            if mode == STATE_STORAGE_SQLITE:
+                return
+
+    if mode == STATE_STORAGE_SQLITE:
+        return
     try:
         with open(filepath, "w", encoding="utf-8") as f:
             _lock_ex(f)
@@ -251,6 +288,7 @@ def update_daily_record(
     )
     history_file = os.path.join(project_root, "data", "daily_records.json")
 
+    mode = get_state_storage_mode()
     data = load_history(history_file)
     if city_name not in data:
         data[city_name] = {}
@@ -359,7 +397,18 @@ def update_daily_record(
         for d in old_dates:
             del data[city][d]
 
-    save_history(history_file, data)
+    if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+        try:
+            _daily_record_repo.upsert_record(city_name, date_str, existing)
+            cutoff = (datetime.now() - timedelta(days=14)).strftime("%Y-%m-%d")
+            _daily_record_repo.delete_older_than(cutoff)
+        except Exception as e:
+            logger.error(f"Error upserting daily record to sqlite city={city_name} date={date_str}: {e}")
+            if mode == STATE_STORAGE_SQLITE:
+                raise
+
+    if mode != STATE_STORAGE_SQLITE:
+        save_history(history_file, data)
 
 
 def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7):

src/analysis/probability_rollout.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any, Dict, List, Optional
+
+
+def _load_json_file(path: str) -> Dict[str, Any]:
+    try:
+        with open(path, "r", encoding="utf-8") as fh:
+            data = json.load(fh)
+        return data if isinstance(data, dict) else {}
+    except Exception:
+        return {}
+
+
+def _sf(value: Any) -> Optional[float]:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except Exception:
+        return None
+
+
+def _append_reason(reasons: List[str], condition: bool, message: str) -> None:
+    if condition:
+        reasons.append(message)
+
+
+def _top_shadow_regressions(by_city: Dict[str, Any], limit: int = 5) -> List[Dict[str, Any]]:
+    rows: List[Dict[str, Any]] = []
+    for city, metrics in (by_city or {}).items():
+        if not isinstance(metrics, dict):
+            continue
+        rows.append(
+            {
+                "city": city,
+                "samples": int(metrics.get("samples") or 0),
+                "delta_mae": _sf(metrics.get("delta_mae")),
+                "delta_bucket_hit_rate": _sf(metrics.get("delta_bucket_hit_rate")),
+                "delta_bucket_brier": _sf(metrics.get("delta_bucket_brier")),
+            }
+        )
+
+    rows.sort(
+        key=lambda row: (
+            -(row["delta_bucket_brier"] or 0.0),
+            row["delta_bucket_hit_rate"] or 0.0,
+            -(row["delta_mae"] or 0.0),
+        )
+    )
+    return rows[:limit]
+
+
+def judge_probability_rollout(
+    evaluation_report: Dict[str, Any],
+    shadow_report: Dict[str, Any],
+) -> Dict[str, Any]:
+    thresholds = {
+        "evaluation_min_samples": 80,
+        "shadow_min_samples": 50,
+        "max_delta_mae": 0.05,
+        "min_delta_crps": -0.02,
+        "min_delta_bucket_hit_rate": 0.0,
+        "max_delta_bucket_brier_promote": 0.02,
+        "max_delta_bucket_brier_observe": 0.15,
+    }
+
+    eval_summary = (evaluation_report or {}).get("summary") or {}
+    eval_delta = eval_summary.get("delta") or {}
+    shadow_summary = (shadow_report or {}).get("summary") or {}
+
+    eval_samples = int(eval_summary.get("sample_count") or 0)
+    shadow_samples = int(shadow_summary.get("samples") or 0)
+    delta_crps = _sf(eval_delta.get("crps"))
+    delta_mae = _sf(eval_delta.get("mae"))
+    delta_hit = _sf(eval_delta.get("bucket_hit_rate"))
+    shadow_delta_mae = _sf(shadow_summary.get("delta_mae"))
+    shadow_delta_hit = _sf(shadow_summary.get("delta_bucket_hit_rate"))
+    shadow_delta_brier = _sf(shadow_summary.get("delta_bucket_brier"))
+
+    promote_reasons: List[str] = []
+    _append_reason(
+        promote_reasons,
+        eval_samples < thresholds["evaluation_min_samples"],
+        f"离线评估样本不足：{eval_samples} < {thresholds['evaluation_min_samples']}",
+    )
+    _append_reason(
+        promote_reasons,
+        shadow_samples < thresholds["shadow_min_samples"],
+        f"shadow 样本不足：{shadow_samples} < {thresholds['shadow_min_samples']}",
+    )
+    _append_reason(
+        promote_reasons,
+        delta_crps is None or delta_crps > thresholds["min_delta_crps"],
+        f"离线 CRPS 改善不足：delta={delta_crps}",
+    )
+    _append_reason(
+        promote_reasons,
+        delta_mae is None or delta_mae > thresholds["max_delta_mae"],
+        f"离线 MAE 退化超限：delta={delta_mae}",
+    )
+    _append_reason(
+        promote_reasons,
+        delta_hit is None or delta_hit < thresholds["min_delta_bucket_hit_rate"],
+        f"离线 bucket 命中率下降：delta={delta_hit}",
+    )
+    _append_reason(
+        promote_reasons,
+        shadow_delta_mae is None or shadow_delta_mae > thresholds["max_delta_mae"],
+        f"shadow MAE 退化超限：delta={shadow_delta_mae}",
+    )
+    _append_reason(
+        promote_reasons,
+        shadow_delta_hit is None or shadow_delta_hit < thresholds["min_delta_bucket_hit_rate"],
+        f"shadow bucket 命中率下降：delta={shadow_delta_hit}",
+    )
+    _append_reason(
+        promote_reasons,
+        shadow_delta_brier is None
+        or shadow_delta_brier > thresholds["max_delta_bucket_brier_promote"],
+        f"shadow bucket brier 退化超限：delta={shadow_delta_brier}",
+    )
+
+    if not promote_reasons:
+        decision = "promote"
+        summary = "离线与 shadow 指标均达标，可以考虑切换 emos_primary。"
+    else:
+        observe_reasons: List[str] = []
+        _append_reason(
+            observe_reasons,
+            eval_samples < thresholds["evaluation_min_samples"],
+            f"离线评估样本不足：{eval_samples}",
+        )
+        _append_reason(
+            observe_reasons,
+            delta_crps is None or delta_crps > thresholds["min_delta_crps"],
+            f"离线 CRPS 改善不足：delta={delta_crps}",
+        )
+        _append_reason(
+            observe_reasons,
+            delta_mae is None or delta_mae > thresholds["max_delta_mae"],
+            f"离线 MAE 退化超限：delta={delta_mae}",
+        )
+        _append_reason(
+            observe_reasons,
+            delta_hit is None or delta_hit < thresholds["min_delta_bucket_hit_rate"],
+            f"离线 bucket 命中率下降：delta={delta_hit}",
+        )
+        _append_reason(
+            observe_reasons,
+            shadow_samples < thresholds["shadow_min_samples"],
+            f"shadow 样本不足：{shadow_samples}",
+        )
+        _append_reason(
+            observe_reasons,
+            shadow_delta_mae is None or shadow_delta_mae > thresholds["max_delta_mae"],
+            f"shadow MAE 退化超限：delta={shadow_delta_mae}",
+        )
+        _append_reason(
+            observe_reasons,
+            shadow_delta_hit is None or shadow_delta_hit < thresholds["min_delta_bucket_hit_rate"],
+            f"shadow bucket 命中率下降：delta={shadow_delta_hit}",
+        )
+        _append_reason(
+            observe_reasons,
+            shadow_delta_brier is None
+            or shadow_delta_brier > thresholds["max_delta_bucket_brier_observe"],
+            f"shadow bucket brier 退化偏大：delta={shadow_delta_brier}",
+        )
+
+        if not observe_reasons:
+            decision = "observe"
+            summary = "离线评估达标，但 shadow 仍需继续观察，暂不切主路径。"
+        else:
+            decision = "hold"
+            summary = "当前指标不足以切换 emos_primary，应继续保持 shadow。"
+
+    return {
+        "decision": decision,
+        "ready_for_primary": decision == "promote",
+        "summary": summary,
+        "thresholds": thresholds,
+        "evaluation": {
+            "sample_count": eval_samples,
+            "delta_crps": delta_crps,
+            "delta_mae": delta_mae,
+            "delta_bucket_hit_rate": delta_hit,
+        },
+        "shadow": {
+            "sample_count": shadow_samples,
+            "delta_mae": shadow_delta_mae,
+            "delta_bucket_hit_rate": shadow_delta_hit,
+            "delta_bucket_brier": shadow_delta_brier,
+        },
+        "blocking_reasons": promote_reasons,
+        "worst_shadow_regressions": _top_shadow_regressions(
+            (shadow_report or {}).get("by_city") or {}
+        ),
+    }
+
+
+def build_rollout_report(
+    evaluation_report_path: str,
+    shadow_report_path: str,
+) -> Dict[str, Any]:
+    evaluation_report = _load_json_file(evaluation_report_path)
+    shadow_report = _load_json_file(shadow_report_path)
+    decision = judge_probability_rollout(evaluation_report, shadow_report)
+    return {
+        "evaluation_report_path": evaluation_report_path,
+        "shadow_report_path": shadow_report_path,
+        "evaluation_report_exists": os.path.exists(evaluation_report_path),
+        "shadow_report_exists": os.path.exists(shadow_report_path),
+        "decision": decision,
+    }

src/analysis/probability_snapshot_archive.py

@@ -5,10 +5,18 @@ import os
 from datetime import datetime
 from typing import Any, Dict, List, Optional
 
+from src.database.runtime_state import (
+    ProbabilitySnapshotRepository,
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+    get_state_storage_mode,
+)
+
 DEDUP_SCAN_LINES = 200
 MU_THRESHOLD = 0.2
 SIGMA_THRESHOLD = 0.15
 MAX_SO_FAR_THRESHOLD = 0.2
+_snapshot_repo = ProbabilitySnapshotRepository()
 
 
 def _sf(value: Any) -> Optional[float]:
@@ -79,7 +87,15 @@ def _load_recent_rows(path: str, max_lines: int = DEDUP_SCAN_LINES) -> List[Dict
 
 
 def _should_skip_append(path: str, payload: Dict[str, Any]) -> bool:
-    recent_rows = _load_recent_rows(path)
+    mode = get_state_storage_mode()
+    if mode == STATE_STORAGE_SQLITE:
+        recent_rows = _snapshot_repo.load_recent_rows(
+            str(payload.get("city") or ""),
+            str(payload.get("date") or ""),
+            DEDUP_SCAN_LINES,
+        )
+    else:
+        recent_rows = _load_recent_rows(path)
     city = payload.get("city")
     date_str = payload.get("date")
     if not city or not date_str:
@@ -183,5 +199,10 @@ def append_probability_snapshot(
     if _should_skip_append(path, payload):
         return
 
-    with open(path, "a", encoding="utf-8") as fh:
-        fh.write(json.dumps(payload, ensure_ascii=False) + "\n")
+    mode = get_state_storage_mode()
+    if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+        _snapshot_repo.append_snapshot(payload)
+
+    if mode != STATE_STORAGE_SQLITE:
+        with open(path, "a", encoding="utf-8") as fh:
+            fh.write(json.dumps(payload, ensure_ascii=False) + "\n")

src/data_collection/metar_sources.py

@@ -8,6 +8,8 @@ from typing import Dict, List, Optional
 import requests
 from loguru import logger
 
+from src.utils.metrics import record_source_call
+
 
 class MetarSourceMixin:
     def get_icao_code(self, city: str) -> Optional[str]:
@@ -24,9 +26,11 @@ class MetarSourceMixin:
         self, city: str, use_fahrenheit: bool = False, utc_offset: int = 0
     ) -> Optional[Dict]:
         """从 NOAA Aviation Weather Center 获取 METAR 航空气象数据。"""
+        started = time.perf_counter()
         icao = self.get_icao_code(city)
         if not icao:
             logger.warning(f"未找到城市 {city} 对应的 ICAO 代码")
+            record_source_call("metar", "current", "missing_icao", (time.perf_counter() - started) * 1000.0)
             return None
 
         cache_key = f"{icao}:{utc_offset}:{use_fahrenheit}"
@@ -35,6 +39,7 @@ class MetarSourceMixin:
             cached = self._metar_cache.get(cache_key)
             if cached and now_ts - cached["t"] < self.metar_cache_ttl_sec:
                 logger.debug(f"METAR cache hit {icao} age={int(now_ts - cached['t'])}s")
+                record_source_call("metar", "current", "cache_hit", (time.perf_counter() - started) * 1000.0)
                 return cached["d"]
 
         try:
@@ -201,6 +206,7 @@ class MetarSourceMixin:
             )
             with self._metar_cache_lock:
                 self._metar_cache[cache_key] = {"d": result, "t": now_ts}
+            record_source_call("metar", "current", "success", (time.perf_counter() - started) * 1000.0)
             return result
 
         except requests.exceptions.RequestException as exc:
@@ -209,10 +215,13 @@ class MetarSourceMixin:
                 stale = self._metar_cache.get(cache_key)
                 if stale:
                     logger.warning(f"METAR {icao} 请求失败，使用缓存回退")
+                    record_source_call("metar", "current", "stale_cache", (time.perf_counter() - started) * 1000.0)
                     return stale["d"]
+            record_source_call("metar", "current", "error", (time.perf_counter() - started) * 1000.0)
             return None
         except (KeyError, IndexError, TypeError) as exc:
             logger.error(f"METAR 数据解析失败 ({icao}): {exc}")
+            record_source_call("metar", "current", "parse_error", (time.perf_counter() - started) * 1000.0)
             return None
 
     def fetch_metar_nearby_cluster(self, icaos: List[str], use_fahrenheit: bool = False) -> list:

src/data_collection/mgm_sources.py

@@ -5,12 +5,15 @@ from typing import Dict, Optional
 
 from loguru import logger
 
+from src.utils.metrics import record_source_call
+
 
 class MgmSourceMixin:
     def fetch_from_mgm(self, istno: str) -> Optional[Dict]:
         """
         从土耳其气象局 (MGM) 获取实时数据和预测 (由用户提供其内部 API)
         """
+        started = datetime.now()
         base_url = "https://servis.mgm.gov.tr/web"
         # 必须带 Origin，否则会被反爬拦截
         headers = {
@@ -202,9 +205,21 @@ class MgmSourceMixin:
                         f"hourly points={len(hourly_rows)}, horizon={horizon_hours:.1f}h"
                     )
 
+            record_source_call(
+                "mgm",
+                "station",
+                "success" if "current" in results else "empty",
+                (datetime.now() - started).total_seconds() * 1000.0,
+            )
             return results if "current" in results else None
         except Exception as e:
             logger.error(f"MGM API 请求失败 ({istno}): {e}")
+            record_source_call(
+                "mgm",
+                "station",
+                "error",
+                (datetime.now() - started).total_seconds() * 1000.0,
+            )
             return None
 
     def fetch_mgm_nearby_stations(self, province: str, root_ist_no: str = None) -> list:
@@ -212,6 +227,7 @@ class MgmSourceMixin:
         获取一个土耳其省份内所有气象站的当前温度及经纬度
         使用多线程辅助抓取，因为直接通过 il={province} 往往只返回 1 个站。
         """
+        started = datetime.now()
         base_url = "https://servis.mgm.gov.tr/web"
         headers = {
             "Origin": "https://www.mgm.gov.tr",
@@ -243,6 +259,12 @@ class MgmSourceMixin:
 
             if not province_ist_nos:
                 logger.warning(f"MGM 找不到省份 {province} 的站点元数据")
+                record_source_call(
+                    "mgm",
+                    "nearby",
+                    "empty",
+                    (datetime.now() - started).total_seconds() * 1000.0,
+                )
                 return []
 
             # 同时确保我们关心的几个核心站一定在里面
@@ -318,8 +340,20 @@ class MgmSourceMixin:
                 })
 
             logger.info(f"📍 MGM 周边测站: 成功并发抓取 {len(results)} 个 {province} 站点的实时气温")
+            record_source_call(
+                "mgm",
+                "nearby",
+                "success" if results else "empty",
+                (datetime.now() - started).total_seconds() * 1000.0,
+            )
             return results
         except Exception as e:
             logger.error(f"Failed to fetch MGM nearby stations for {province}: {e}")
+            record_source_call(
+                "mgm",
+                "nearby",
+                "error",
+                (datetime.now() - started).total_seconds() * 1000.0,
+            )
             return []
 

src/data_collection/nws_open_meteo_sources.py

@@ -6,6 +6,8 @@ from typing import Dict, Optional
 
 from loguru import logger
 
+from src.utils.metrics import record_source_call
+
 
 class NwsOpenMeteoSourceMixin:
     def fetch_nws(self, lat: float, lon: float) -> Optional[Dict]:
@@ -13,6 +15,7 @@ class NwsOpenMeteoSourceMixin:
         从 NWS (美国国家气象局) 获取高精度预报
         仅适用于美国城市，全球 VPS 均可访问
         """
+        started = time.perf_counter()
         try:
             # 1. 获取网格点
             points_url = f"https://api.weather.gov/points/{lat},{lon}"
@@ -28,6 +31,7 @@ class NwsOpenMeteoSourceMixin:
             forecast_url = properties.get("forecast")
             hourly_url = properties.get("forecastHourly")
             if not forecast_url:
+                record_source_call("nws", "forecast", "empty", (time.perf_counter() - started) * 1000.0)
                 return None
 
             # 2. 获取预报
@@ -39,6 +43,7 @@ class NwsOpenMeteoSourceMixin:
 
             periods = forecast_data.get("properties", {}).get("periods", [])
             if not periods:
+                record_source_call("nws", "forecast", "empty", (time.perf_counter() - started) * 1000.0)
                 return None
 
             hourly_periods = []
@@ -89,7 +94,7 @@ class NwsOpenMeteoSourceMixin:
                         today_high = p.get("temperature")
                         break
 
-            return {
+            result = {
                 "source": "nws",
                 "today_high": today_high,
                 "unit": "fahrenheit",
@@ -124,8 +129,11 @@ class NwsOpenMeteoSourceMixin:
                 ],
                 "active_alerts": active_alerts,
             }
+            record_source_call("nws", "forecast", "success", (time.perf_counter() - started) * 1000.0)
+            return result
         except Exception as e:
             logger.warning(f"NWS 请求失败: {e}")
+            record_source_call("nws", "forecast", "error", (time.perf_counter() - started) * 1000.0)
             return None
 
     def fetch_from_open_meteo(
@@ -144,6 +152,7 @@ class NwsOpenMeteoSourceMixin:
             forecast_days: Number of forecast days to fetch (default 14 to cover all market dates)
             use_fahrenheit: Whether to return temperatures in Fahrenheit (for US markets)
         """
+        started = time.perf_counter()
         cache_key = (
             f"{round(float(lat), 4)}:{round(float(lon), 4)}:"
             f"{forecast_days}:{'f' if use_fahrenheit else 'c'}"
@@ -156,9 +165,11 @@ class NwsOpenMeteoSourceMixin:
                 remaining = int(self._open_meteo_rate_limit_until - now_ts)
                 logger.debug(f"Open-Meteo 冷却期中，跳过请求，还需 {remaining}s")
                 with self._open_meteo_cache_lock:
-                    stale = self._open_meteo_cache.get(cache_key)
-                    if stale and isinstance(stale.get("data"), dict):
-                        return dict(stale["data"])
+                        stale = self._open_meteo_cache.get(cache_key)
+                        if stale and isinstance(stale.get("data"), dict):
+                            record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0)
+                            return dict(stale["data"])
+                record_source_call("open_meteo", "forecast", "cooldown_skip", (time.perf_counter() - started) * 1000.0)
                 return None
         with self._open_meteo_cache_lock:
             cached = self._open_meteo_cache.get(cache_key)
@@ -168,6 +179,7 @@ class NwsOpenMeteoSourceMixin:
             ):
                 cached_data = cached.get("data")
                 if isinstance(cached_data, dict):
+                    record_source_call("open_meteo", "forecast", "cache_hit", (time.perf_counter() - started) * 1000.0)
                     return dict(cached_data)
         try:
             url = "https://api.open-meteo.com/v1/forecast"
@@ -259,6 +271,7 @@ class NwsOpenMeteoSourceMixin:
                     "data": dict(result),
                 }
             self._flush_open_meteo_disk_cache()
+            record_source_call("open_meteo", "forecast", "success", (time.perf_counter() - started) * 1000.0)
             return result
         except Exception as e:
             status_code = getattr(getattr(e, "response", None), "status_code", None)
@@ -287,7 +300,9 @@ class NwsOpenMeteoSourceMixin:
                 if stale and isinstance(stale.get("data"), dict):
                     fallback = dict(stale["data"])
                     fallback["stale_cache"] = True
+                    record_source_call("open_meteo", "forecast", "stale_cache", (time.perf_counter() - started) * 1000.0)
                     return fallback
+            record_source_call("open_meteo", "forecast", "error", (time.perf_counter() - started) * 1000.0)
             return None
 
     def fetch_ensemble(
@@ -300,6 +315,7 @@ class NwsOpenMeteoSourceMixin:
         从 Open-Meteo Ensemble API 获取 51 成员集合预报
         用于计算预报不确定性范围（散度）
         """
+        started = time.perf_counter()
         cache_key = (
             f"{round(float(lat), 4)}:{round(float(lon), 4)}:"
             f"{'f' if use_fahrenheit else 'c'}"
@@ -314,7 +330,9 @@ class NwsOpenMeteoSourceMixin:
                 with self._ensemble_cache_lock:
                     stale = self._ensemble_cache.get(cache_key)
                     if stale and isinstance(stale.get("data"), dict):
+                        record_source_call("open_meteo", "ensemble", "stale_cache", (time.perf_counter() - started) * 1000.0)
                         return dict(stale["data"])
+                record_source_call("open_meteo", "ensemble", "cooldown_skip", (time.perf_counter() - started) * 1000.0)
                 return None
                 
         with self._ensemble_cache_lock:
@@ -326,6 +344,7 @@ class NwsOpenMeteoSourceMixin:
             ):
                 cached_data = cached.get("data")
                 if isinstance(cached_data, dict):
+                    record_source_call("open_meteo", "ensemble", "cache_hit", (time.perf_counter() - started) * 1000.0)
                     return dict(cached_data)
         try:
             url = "https://ensemble-api.open-meteo.com/v1/ensemble"
@@ -371,6 +390,7 @@ class NwsOpenMeteoSourceMixin:
 
             if len(today_highs) < 3:
                 logger.warning(f"Ensemble 数据不足: 仅获取 {len(today_highs)} 个成员")
+                record_source_call("open_meteo", "ensemble", "empty", (time.perf_counter() - started) * 1000.0)
                 return None
 
             today_highs.sort()
@@ -400,6 +420,7 @@ class NwsOpenMeteoSourceMixin:
                     "data": dict(result),
                 }
             self._flush_open_meteo_disk_cache()
+            record_source_call("open_meteo", "ensemble", "success", (time.perf_counter() - started) * 1000.0)
             return result
         except Exception as e:
             status_code = getattr(getattr(e, "response", None), "status_code", None)
@@ -425,7 +446,9 @@ class NwsOpenMeteoSourceMixin:
                 if stale and isinstance(stale.get("data"), dict):
                     fallback = dict(stale["data"])
                     fallback["stale_cache"] = True
+                    record_source_call("open_meteo", "ensemble", "stale_cache", (time.perf_counter() - started) * 1000.0)
                     return fallback
+            record_source_call("open_meteo", "ensemble", "error", (time.perf_counter() - started) * 1000.0)
             return None
 
     def fetch_multi_model(
@@ -448,6 +471,7 @@ class NwsOpenMeteoSourceMixin:
 
         返回 3 天的预报数据，支持今日+明日共识分析
         """
+        started = time.perf_counter()
         cache_city = str(city or "").strip().lower()
         cache_key = (
             f"{round(float(lat), 4)}:{round(float(lon), 4)}:{cache_city}:"
@@ -463,7 +487,9 @@ class NwsOpenMeteoSourceMixin:
                 with self._multi_model_cache_lock:
                     stale = self._multi_model_cache.get(cache_key)
                     if stale and isinstance(stale.get("data"), dict):
+                        record_source_call("open_meteo", "multi_model", "stale_cache", (time.perf_counter() - started) * 1000.0)
                         return dict(stale["data"])
+                record_source_call("open_meteo", "multi_model", "cooldown_skip", (time.perf_counter() - started) * 1000.0)
                 return None
 
         with self._multi_model_cache_lock:
@@ -475,6 +501,7 @@ class NwsOpenMeteoSourceMixin:
             ):
                 cached_data = cached.get("data")
                 if isinstance(cached_data, dict):
+                    record_source_call("open_meteo", "multi_model", "cache_hit", (time.perf_counter() - started) * 1000.0)
                     return dict(cached_data)
         try:
             url = "https://api.open-meteo.com/v1/forecast"
@@ -524,6 +551,7 @@ class NwsOpenMeteoSourceMixin:
 
             if not daily_forecasts:
                 logger.warning("Multi-model: 无有效模型数据")
+                record_source_call("open_meteo", "multi_model", "empty", (time.perf_counter() - started) * 1000.0)
                 return None
 
             # 今天的预报 (向后兼容)
@@ -548,6 +576,7 @@ class NwsOpenMeteoSourceMixin:
                     "data": dict(result),
                 }
             self._flush_open_meteo_disk_cache()
+            record_source_call("open_meteo", "multi_model", "success", (time.perf_counter() - started) * 1000.0)
             return result
         except Exception as e:
             status_code = getattr(getattr(e, "response", None), "status_code", None)
@@ -573,6 +602,8 @@ class NwsOpenMeteoSourceMixin:
                 if stale and isinstance(stale.get("data"), dict):
                     fallback = dict(stale["data"])
                     fallback["stale_cache"] = True
+                    record_source_call("open_meteo", "multi_model", "stale_cache", (time.perf_counter() - started) * 1000.0)
                     return fallback
+            record_source_call("open_meteo", "multi_model", "error", (time.perf_counter() - started) * 1000.0)
             return None
 

src/data_collection/open_meteo_cache.py

@@ -5,11 +5,30 @@ import os
 import time
 
 from loguru import logger
+from src.database.runtime_state import (
+    OpenMeteoCacheRepository,
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+    get_state_storage_mode,
+)
+
+_open_meteo_cache_repo = OpenMeteoCacheRepository()
 
 
 class OpenMeteoCacheMixin:
     def _load_open_meteo_disk_cache(self) -> None:
         """启动时从磁盘加载 Open-Meteo 三类缓存，避免重启后冷启动打爆 API"""
+        mode = get_state_storage_mode()
+        if mode == STATE_STORAGE_SQLITE:
+            try:
+                saved = _open_meteo_cache_repo.load_payload(self._disk_cache_max_age_sec)
+                loaded = self._merge_open_meteo_payload(saved)
+                self._disk_cache_last_mtime = _open_meteo_cache_repo.latest_updated_at()
+                if loaded:
+                    logger.info(f"✅ 从 SQLite 加载 Open-Meteo 缓存 {loaded} 条")
+                return
+            except Exception as exc:
+                logger.error(f"SQLite Open-Meteo 缓存加载失败，fallback file: {exc}")
         try:
             path = self._disk_cache_path
             if not os.path.exists(path):
@@ -58,11 +77,26 @@ class OpenMeteoCacheMixin:
             self._disk_cache_last_mtime = current_mtime
             if loaded:
                 logger.info(f"✅ 从磁盘加载 Open-Meteo 缓存 {loaded} 条 ({self._disk_cache_path})")
+            if mode == STATE_STORAGE_DUAL:
+                try:
+                    _open_meteo_cache_repo.replace_payload(saved, self._disk_cache_max_age_sec)
+                except Exception as exc:
+                    logger.warning(f"SQLite Open-Meteo 缓存同步失败: {exc}")
         except Exception as exc:
             logger.warning(f"磁盘缓存加载失败（首次启动不影响运行）: {exc}")
 
     def _maybe_reload_open_meteo_disk_cache(self) -> None:
         """跨进程共享缓存：当缓存文件有更新时增量重载到当前进程内存"""
+        mode = get_state_storage_mode()
+        if mode == STATE_STORAGE_SQLITE:
+            try:
+                latest_updated_at = _open_meteo_cache_repo.latest_updated_at()
+                if latest_updated_at <= self._disk_cache_last_mtime:
+                    return
+                self._load_open_meteo_disk_cache()
+            except Exception:
+                pass
+            return
         try:
             path = self._disk_cache_path
             if not os.path.exists(path):
@@ -76,8 +110,8 @@ class OpenMeteoCacheMixin:
 
     def _flush_open_meteo_disk_cache(self) -> None:
         """将三类 Open-Meteo 内存缓存持久化到磁盘"""
+        mode = get_state_storage_mode()
         try:
-            os.makedirs(os.path.dirname(self._disk_cache_path), exist_ok=True)
             with self._open_meteo_cache_lock:
                 forecast_snapshot = dict(self._open_meteo_cache)
             with self._ensemble_cache_lock:
@@ -90,15 +124,50 @@ class OpenMeteoCacheMixin:
                 "multi_model": multi_model_snapshot,
                 "saved_at": time.time(),
             }
-            with self._disk_cache_lock:
-                tmp_path = self._disk_cache_path + ".tmp"
-                with open(tmp_path, "w", encoding="utf-8") as f:
-                    json.dump(payload, f)
-                os.replace(tmp_path, self._disk_cache_path)
-            self._disk_cache_last_mtime = os.path.getmtime(self._disk_cache_path)
+            if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+                _open_meteo_cache_repo.replace_payload(payload, self._disk_cache_max_age_sec)
+                self._disk_cache_last_mtime = _open_meteo_cache_repo.latest_updated_at()
+            if mode != STATE_STORAGE_SQLITE:
+                os.makedirs(os.path.dirname(self._disk_cache_path), exist_ok=True)
+                with self._disk_cache_lock:
+                    tmp_path = self._disk_cache_path + ".tmp"
+                    with open(tmp_path, "w", encoding="utf-8") as f:
+                        json.dump(payload, f)
+                    os.replace(tmp_path, self._disk_cache_path)
+                self._disk_cache_last_mtime = max(
+                    self._disk_cache_last_mtime,
+                    os.path.getmtime(self._disk_cache_path),
+                )
         except Exception as exc:
             logger.warning(f"磁盘缓存写入失败: {exc}")
 
+    def _merge_open_meteo_payload(self, saved: dict) -> int:
+        now = time.time()
+        max_age = max(600, self._disk_cache_max_age_sec)
+        loaded = 0
+        with self._open_meteo_cache_lock:
+            for key, entry in (saved.get("forecast", {}) or {}).items():
+                if now - float(entry.get("t", 0)) < max_age:
+                    old = self._open_meteo_cache.get(key)
+                    if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)):
+                        self._open_meteo_cache[key] = entry
+                        loaded += 1
+        with self._ensemble_cache_lock:
+            for key, entry in (saved.get("ensemble", {}) or {}).items():
+                if now - float(entry.get("t", 0)) < max_age:
+                    old = self._ensemble_cache.get(key)
+                    if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)):
+                        self._ensemble_cache[key] = entry
+                        loaded += 1
+        with self._multi_model_cache_lock:
+            for key, entry in (saved.get("multi_model", {}) or {}).items():
+                if now - float(entry.get("t", 0)) < max_age:
+                    old = self._multi_model_cache.get(key)
+                    if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)):
+                        self._multi_model_cache[key] = entry
+                        loaded += 1
+        return loaded
+
     def _wait_open_meteo_slot(self, endpoint: str) -> None:
         """Simple per-process rate gate for Open-Meteo endpoints."""
         min_interval = self._open_meteo_min_interval_sec

src/database/runtime_state.py

@@ -0,0 +1,500 @@
+
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import threading
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from loguru import logger
+
+from src.database.db_manager import DBManager
+
+STATE_STORAGE_FILE = "file"
+STATE_STORAGE_DUAL = "dual"
+STATE_STORAGE_SQLITE = "sqlite"
+VALID_STATE_STORAGE_MODES = {
+    STATE_STORAGE_FILE,
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+}
+
+_LOGGED_MODES: set[str] = set()
+
+
+def get_state_storage_mode() -> str:
+    raw = str(os.getenv("POLYWEATHER_STATE_STORAGE_MODE") or STATE_STORAGE_DUAL).strip().lower()
+    if raw not in VALID_STATE_STORAGE_MODES:
+        logger.warning(
+            f"invalid POLYWEATHER_STATE_STORAGE_MODE={raw!r}, fallback to {STATE_STORAGE_DUAL}"
+        )
+        raw = STATE_STORAGE_DUAL
+    if raw not in _LOGGED_MODES:
+        logger.info(f"runtime state storage mode={raw}")
+        _LOGGED_MODES.add(raw)
+    return raw
+
+
+class RuntimeStateDB:
+    _instance: Optional["RuntimeStateDB"] = None
+    _instance_lock = threading.Lock()
+
+    def __init__(self, db_path: Optional[str] = None):
+        self.db_path = DBManager(db_path).db_path
+        self._init_tables()
+
+    @classmethod
+    def instance(cls) -> "RuntimeStateDB":
+        with cls._instance_lock:
+            if cls._instance is None:
+                cls._instance = cls()
+            return cls._instance
+
+    def connect(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _init_tables(self) -> None:
+        with self.connect() as conn:
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS daily_records_store (
+                    city TEXT NOT NULL,
+                    target_date TEXT NOT NULL,
+                    actual_high REAL,
+                    deb_prediction REAL,
+                    mu REAL,
+                    updated_at REAL NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    PRIMARY KEY (city, target_date)
+                )
+                """
+            )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS telegram_alert_last_by_city (
+                    city TEXT PRIMARY KEY,
+                    signature TEXT,
+                    trigger_key TEXT,
+                    severity TEXT,
+                    ts INTEGER,
+                    active INTEGER DEFAULT 0,
+                    cleared_ts INTEGER,
+                    evidence_json TEXT
+                )
+                """
+            )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS telegram_alert_signature_state (
+                    signature TEXT PRIMARY KEY,
+                    ts INTEGER NOT NULL
+                )
+                """
+            )
+            conn.execute(
+                """
+                CREATE TABLE IF NOT EXISTS probability_training_snapshots_store (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    city TEXT NOT NULL,
+                    target_date TEXT NOT NULL,
+                    timestamp TEXT NOT NULL,
+                    raw_mu REAL,
+                    raw_sigma REAL,
+                    max_so_far REAL,
+                    peak_status TEXT,
+                    probability_mode TEXT,
+                    legacy_top_bucket INTEGER,
+                    shadow_top_bucket INTEGER,
+                    payload_json TEXT NOT NULL
+                )
+                """
+            )
+            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 open_meteo_cache_store (
+                    source_kind TEXT NOT NULL,
+                    cache_key TEXT NOT NULL,
+                    updated_at REAL NOT NULL,
+                    expires_at REAL,
+                    payload_json TEXT NOT NULL,
+                    PRIMARY KEY (source_kind, cache_key)
+                )
+                """
+            )
+            conn.execute(
+                "CREATE INDEX IF NOT EXISTS idx_open_meteo_cache_expires ON open_meteo_cache_store(source_kind, expires_at)"
+            )
+            conn.commit()
+
+
+class DailyRecordRepository:
+    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, payload_json FROM daily_records_store ORDER BY city, target_date"
+            ).fetchall()
+        for row in rows:
+            try:
+                payload = json.loads(row["payload_json"])
+            except Exception:
+                continue
+            city = str(row["city"])
+            date_str = str(row["target_date"])
+            out.setdefault(city, {})[date_str] = payload
+        return out
+
+    def upsert_record(self, city: str, target_date: str, record: Dict[str, Any]) -> None:
+        payload_json = json.dumps(record, ensure_ascii=False)
+        updated_at = time.time()
+        with self.db.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO daily_records_store (
+                    city, target_date, actual_high, deb_prediction, mu, updated_at, payload_json
+                ) VALUES (?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(city, target_date) DO UPDATE SET
+                    actual_high = excluded.actual_high,
+                    deb_prediction = excluded.deb_prediction,
+                    mu = excluded.mu,
+                    updated_at = excluded.updated_at,
+                    payload_json = excluded.payload_json
+                """,
+                (
+                    city,
+                    target_date,
+                    record.get("actual_high"),
+                    record.get("deb_prediction"),
+                    record.get("mu"),
+                    updated_at,
+                    payload_json,
+                ),
+            )
+            conn.commit()
+
+    def replace_all(self, data: Dict[str, Dict[str, Dict[str, Any]]]) -> int:
+        count = 0
+        with self.db.connect() as conn:
+            conn.execute("DELETE FROM daily_records_store")
+            for city, city_rows in (data or {}).items():
+                if not isinstance(city_rows, dict):
+                    continue
+                for target_date, record in city_rows.items():
+                    payload_json = json.dumps(record, ensure_ascii=False)
+                    conn.execute(
+                        """
+                        INSERT INTO daily_records_store (
+                            city, target_date, actual_high, deb_prediction, mu, updated_at, payload_json
+                        ) VALUES (?, ?, ?, ?, ?, ?, ?)
+                        """,
+                        (
+                            city,
+                            target_date,
+                            record.get("actual_high"),
+                            record.get("deb_prediction"),
+                            record.get("mu"),
+                            time.time(),
+                            payload_json,
+                        ),
+                    )
+                    count += 1
+            conn.commit()
+        return count
+
+    def delete_older_than(self, cutoff_date: str) -> int:
+        with self.db.connect() as conn:
+            cur = conn.execute(
+                "DELETE FROM daily_records_store WHERE target_date < ?",
+                (cutoff_date,),
+            )
+            conn.commit()
+            return int(cur.rowcount or 0)
+
+
+class TelegramAlertStateRepository:
+    def __init__(self, db: Optional[RuntimeStateDB] = None):
+        self.db = db or RuntimeStateDB.instance()
+
+    def load_state(self) -> Dict[str, Any]:
+        state = {"last_by_city": {}, "by_signature": {}}
+        with self.db.connect() as conn:
+            city_rows = conn.execute(
+                "SELECT city, signature, trigger_key, severity, ts, active, cleared_ts, evidence_json FROM telegram_alert_last_by_city"
+            ).fetchall()
+            sig_rows = conn.execute(
+                "SELECT signature, ts FROM telegram_alert_signature_state"
+            ).fetchall()
+        for row in city_rows:
+            entry = {
+                "signature": row["signature"],
+                "trigger_key": row["trigger_key"],
+                "severity": row["severity"],
+                "ts": row["ts"],
+                "active": bool(row["active"]),
+            }
+            if row["cleared_ts"] is not None:
+                entry["cleared_ts"] = row["cleared_ts"]
+            if row["evidence_json"]:
+                try:
+                    entry["evidence"] = json.loads(row["evidence_json"])
+                except Exception:
+                    pass
+            state["last_by_city"][str(row["city"])] = entry
+        for row in sig_rows:
+            state["by_signature"][str(row["signature"])] = int(row["ts"] or 0)
+        return state
+
+    def save_state(self, state: Dict[str, Any]) -> None:
+        last_by_city = state.get("last_by_city") or {}
+        by_signature = state.get("by_signature") or {}
+        with self.db.connect() as conn:
+            conn.execute("DELETE FROM telegram_alert_last_by_city")
+            conn.execute("DELETE FROM telegram_alert_signature_state")
+            for city, row in last_by_city.items():
+                if not isinstance(row, dict):
+                    continue
+                conn.execute(
+                    """
+                    INSERT INTO telegram_alert_last_by_city (
+                        city, signature, trigger_key, severity, ts, active, cleared_ts, evidence_json
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (
+                        city,
+                        row.get("signature"),
+                        row.get("trigger_key"),
+                        row.get("severity"),
+                        int(row.get("ts") or 0),
+                        1 if row.get("active") else 0,
+                        row.get("cleared_ts"),
+                        json.dumps(row.get("evidence"), ensure_ascii=False)
+                        if row.get("evidence") is not None
+                        else None,
+                    ),
+                )
+            for signature, ts in by_signature.items():
+                conn.execute(
+                    "INSERT INTO telegram_alert_signature_state (signature, ts) VALUES (?, ?)",
+                    (signature, int(ts or 0)),
+                )
+            conn.commit()
+
+    def replace_from_state(self, state: Dict[str, Any]) -> int:
+        self.save_state(state)
+        return len((state.get("last_by_city") or {})) + len((state.get("by_signature") or {}))
+
+
+class ProbabilitySnapshotRepository:
+    def __init__(self, db: Optional[RuntimeStateDB] = None):
+        self.db = db or RuntimeStateDB.instance()
+
+    def append_snapshot(self, payload: Dict[str, Any]) -> None:
+        legacy_top = _top_bucket(payload.get("prob_snapshot"))
+        shadow_top = _top_bucket(payload.get("shadow_prob_snapshot"))
+        with self.db.connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO probability_training_snapshots_store (
+                    city, target_date, timestamp, raw_mu, raw_sigma, max_so_far,
+                    peak_status, probability_mode, legacy_top_bucket, shadow_top_bucket, payload_json
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    payload.get("city"),
+                    payload.get("date"),
+                    payload.get("timestamp"),
+                    payload.get("raw_mu"),
+                    payload.get("raw_sigma"),
+                    payload.get("max_so_far"),
+                    payload.get("peak_status"),
+                    payload.get("probability_mode"),
+                    legacy_top,
+                    shadow_top,
+                    json.dumps(payload, ensure_ascii=False),
+                ),
+            )
+            conn.commit()
+
+    def load_recent_rows(self, city: str, target_date: str, limit: int) -> List[Dict[str, Any]]:
+        with self.db.connect() as conn:
+            rows = conn.execute(
+                """
+                SELECT payload_json
+                FROM probability_training_snapshots_store
+                WHERE city = ? AND target_date = ?
+                ORDER BY id DESC
+                LIMIT ?
+                """,
+                (city, target_date, int(limit)),
+            ).fetchall()
+        out = []
+        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 probability_training_snapshots_store ORDER BY id"
+            ).fetchall()
+        out = []
+        for row in rows:
+            try:
+                out.append(json.loads(row["payload_json"]))
+            except Exception:
+                continue
+        return out
+
+    def replace_all(self, rows: List[Dict[str, Any]]) -> int:
+        count = 0
+        with self.db.connect() as conn:
+            conn.execute("DELETE FROM probability_training_snapshots_store")
+            for payload in rows or []:
+                if not isinstance(payload, dict):
+                    continue
+                legacy_top = _top_bucket(payload.get("prob_snapshot"))
+                shadow_top = _top_bucket(payload.get("shadow_prob_snapshot"))
+                conn.execute(
+                    """
+                    INSERT INTO probability_training_snapshots_store (
+                        city, target_date, timestamp, raw_mu, raw_sigma, max_so_far,
+                        peak_status, probability_mode, legacy_top_bucket, shadow_top_bucket, payload_json
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (
+                        payload.get("city"),
+                        payload.get("date"),
+                        payload.get("timestamp"),
+                        payload.get("raw_mu"),
+                        payload.get("raw_sigma"),
+                        payload.get("max_so_far"),
+                        payload.get("peak_status"),
+                        payload.get("probability_mode"),
+                        legacy_top,
+                        shadow_top,
+                        json.dumps(payload, ensure_ascii=False),
+                    ),
+                )
+                count += 1
+            conn.commit()
+        return count
+
+
+class OpenMeteoCacheRepository:
+    def __init__(self, db: Optional[RuntimeStateDB] = None):
+        self.db = db or RuntimeStateDB.instance()
+
+    def replace_payload(self, payload: Dict[str, Any], max_age: int) -> int:
+        count = 0
+        now = time.time()
+        with self.db.connect() as conn:
+            conn.execute("DELETE FROM open_meteo_cache_store")
+            for source_kind in ("forecast", "ensemble", "multi_model"):
+                bucket = payload.get(source_kind) or {}
+                if not isinstance(bucket, dict):
+                    continue
+                for cache_key, entry in bucket.items():
+                    if not isinstance(entry, dict):
+                        continue
+                    updated_at = float(entry.get("t") or now)
+                    expires_at = updated_at + max_age
+                    conn.execute(
+                        """
+                        INSERT INTO open_meteo_cache_store (
+                            source_kind, cache_key, updated_at, expires_at, payload_json
+                        ) VALUES (?, ?, ?, ?, ?)
+                        """,
+                        (
+                            source_kind,
+                            cache_key,
+                            updated_at,
+                            expires_at,
+                            json.dumps(entry, ensure_ascii=False),
+                        ),
+                    )
+                    count += 1
+            conn.commit()
+        return count
+
+    def load_payload(self, max_age: int) -> Dict[str, Any]:
+        now = time.time()
+        payload: Dict[str, Any] = {
+            "forecast": {},
+            "ensemble": {},
+            "multi_model": {},
+            "saved_at": now,
+        }
+        with self.db.connect() as conn:
+            rows = conn.execute(
+                "SELECT source_kind, cache_key, updated_at, payload_json FROM open_meteo_cache_store"
+            ).fetchall()
+        for row in rows:
+            updated_at = float(row["updated_at"] or 0)
+            if now - updated_at >= max(600, max_age):
+                continue
+            try:
+                entry = json.loads(row["payload_json"])
+            except Exception:
+                continue
+            payload.setdefault(str(row["source_kind"]), {})[str(row["cache_key"])] = entry
+        return payload
+
+    def latest_updated_at(self) -> float:
+        with self.db.connect() as conn:
+            row = conn.execute(
+                "SELECT MAX(updated_at) AS max_updated_at FROM open_meteo_cache_store"
+            ).fetchone()
+        if not row:
+            return 0.0
+        try:
+            return float(row["max_updated_at"] or 0.0)
+        except Exception:
+            return 0.0
+
+
+def _top_bucket(snapshot: Optional[List[Dict[str, Any]]]) -> Optional[int]:
+    best_value = None
+    best_prob = -1.0
+    for row in snapshot or []:
+        if not isinstance(row, dict):
+            continue
+        value = row.get("v")
+        if value is None:
+            value = row.get("value")
+        try:
+            ivalue = int(value)
+        except Exception:
+            continue
+        prob = row.get("p")
+        if prob is None:
+            prob = row.get("probability")
+        try:
+            fprob = float(prob)
+        except Exception:
+            continue
+        if fprob > best_prob:
+            best_prob = fprob
+            best_value = ivalue
+    return best_value
+
+
+def get_runtime_data_dir() -> str:
+    raw = str(os.getenv("POLYWEATHER_RUNTIME_DATA_DIR") or "").strip()
+    if raw:
+        return raw
+    project_root = Path(__file__).resolve().parents[2]
+    return str(project_root / "data")

src/utils/metrics.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import threading
+from typing import Dict, Iterable, List, Optional, Tuple
+
+
+LabelTuple = Tuple[Tuple[str, str], ...]
+
+
+class _MetricsRegistry:
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._counters: Dict[Tuple[str, LabelTuple], float] = {}
+        self._gauges: Dict[Tuple[str, LabelTuple], float] = {}
+        self._histograms: Dict[Tuple[str, LabelTuple], Dict[str, float]] = {}
+
+    @staticmethod
+    def _normalize_labels(labels: Dict[str, object]) -> LabelTuple:
+        return tuple(
+            sorted((str(key), str(value)) for key, value in labels.items() if value is not None)
+        )
+
+    def inc_counter(self, name: str, amount: float = 1.0, **labels: object) -> None:
+        key = (name, self._normalize_labels(labels))
+        with self._lock:
+            self._counters[key] = self._counters.get(key, 0.0) + amount
+
+    def set_gauge(self, name: str, value: float, **labels: object) -> None:
+        key = (name, self._normalize_labels(labels))
+        with self._lock:
+            self._gauges[key] = value
+
+    def observe(self, name: str, value: float, **labels: object) -> None:
+        key = (name, self._normalize_labels(labels))
+        with self._lock:
+            bucket = self._histograms.setdefault(
+                key,
+                {"count": 0.0, "sum": 0.0, "max": 0.0},
+            )
+            bucket["count"] += 1.0
+            bucket["sum"] += value
+            bucket["max"] = max(bucket["max"], value)
+
+    def snapshot(self) -> Dict[str, object]:
+        with self._lock:
+            return {
+                "counters": dict(self._counters),
+                "gauges": dict(self._gauges),
+                "histograms": {
+                    key: dict(value) for key, value in self._histograms.items()
+                },
+            }
+
+    def export_prometheus(self) -> str:
+        snap = self.snapshot()
+        lines: List[str] = []
+        for name, labels, value in _iter_metrics(snap["counters"]):
+            lines.append(_prom_line(name, value, labels))
+        for name, labels, value in _iter_metrics(snap["gauges"]):
+            lines.append(_prom_line(name, value, labels))
+        for (name, labels), stats in sorted(snap["histograms"].items()):
+            lines.append(_prom_line(f"{name}_count", stats["count"], labels))
+            lines.append(_prom_line(f"{name}_sum", stats["sum"], labels))
+            lines.append(_prom_line(f"{name}_max", stats["max"], labels))
+        return "\n".join(lines) + ("\n" if lines else "")
+
+
+def _iter_metrics(entries: Dict[Tuple[str, LabelTuple], float]) -> Iterable[Tuple[str, LabelTuple, float]]:
+    for (name, labels), value in sorted(entries.items()):
+        yield name, labels, value
+
+
+def _prom_line(name: str, value: float, labels: LabelTuple) -> str:
+    if labels:
+        def _escape(label_value: str) -> str:
+            return label_value.replace("\\", "\\\\").replace('"', '\\"')
+
+        label_str = ",".join(
+            f'{key}="{_escape(str(val))}"'
+            for key, val in labels
+        )
+        return f"{name}{{{label_str}}} {value}"
+    return f"{name} {value}"
+
+
+METRICS = _MetricsRegistry()
+
+
+def counter_inc(name: str, amount: float = 1.0, **labels: object) -> None:
+    METRICS.inc_counter(name, amount=amount, **labels)
+
+
+def gauge_set(name: str, value: float, **labels: object) -> None:
+    METRICS.set_gauge(name, value=value, **labels)
+
+
+def histogram_observe(name: str, value: float, **labels: object) -> None:
+    METRICS.observe(name, value=value, **labels)
+
+
+def record_source_call(source: str, operation: str, outcome: str, duration_ms: Optional[float] = None) -> None:
+    counter_inc(
+        "polyweather_source_requests_total",
+        source=source,
+        operation=operation,
+        outcome=outcome,
+    )
+    if duration_ms is not None:
+        histogram_observe(
+            "polyweather_source_request_duration_ms",
+            duration_ms,
+            source=source,
+            operation=operation,
+            outcome=outcome,
+        )
+
+
+def build_metrics_summary() -> Dict[str, object]:
+    snapshot = METRICS.snapshot()
+    request_total = 0.0
+    source_total = 0.0
+    source_errors = 0.0
+    for (name, labels), value in snapshot["counters"].items():
+        if name == "polyweather_http_requests_total":
+            request_total += value
+        if name == "polyweather_source_requests_total":
+            source_total += value
+            label_map = dict(labels)
+            if label_map.get("outcome") not in {"success", "cache_hit"}:
+                source_errors += value
+    return {
+        "http_requests_total": int(request_total),
+        "source_requests_total": int(source_total),
+        "source_error_total": int(source_errors),
+    }
+
+
+def export_prometheus_metrics() -> str:
+    return METRICS.export_prometheus()

src/utils/telegram_push.py

@@ -9,6 +9,12 @@ from typing import Any, Dict, List, Optional, Tuple
 
 from loguru import logger
 
+from src.database.runtime_state import (
+    STATE_STORAGE_DUAL,
+    STATE_STORAGE_SQLITE,
+    TelegramAlertStateRepository,
+    get_state_storage_mode,
+)
 from src.data_collection.city_registry import CITY_REGISTRY
 from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env
 
@@ -19,6 +25,7 @@ SEVERITY_RANK = {
     "medium": 2,
     "high": 3,
 }
+_telegram_state_repo = TelegramAlertStateRepository()
 
 
 def _env_bool(name: str, default: bool) -> bool:
@@ -184,7 +191,18 @@ def _state_file() -> str:
 
 
 def _load_state(path: str) -> Dict[str, Any]:
+    mode = get_state_storage_mode()
+    if mode == STATE_STORAGE_SQLITE:
+        try:
+            return _telegram_state_repo.load_state()
+        except Exception as exc:
+            logger.error(f"failed to load telegram push state from sqlite: {exc}")
     if not os.path.exists(path):
+        if mode == STATE_STORAGE_DUAL:
+            try:
+                return _telegram_state_repo.load_state()
+            except Exception:
+                return {"last_by_city": {}, "by_signature": {}}
         return {"last_by_city": {}, "by_signature": {}}
     try:
         with open(path, "r", encoding="utf-8") as fh:
@@ -199,6 +217,11 @@ def _load_state(path: str) -> Dict[str, Any]:
 
 
 def _save_state(path: str, state: Dict[str, Any]) -> None:
+    mode = get_state_storage_mode()
+    if mode in {STATE_STORAGE_DUAL, STATE_STORAGE_SQLITE}:
+        _telegram_state_repo.save_state(state)
+    if mode == STATE_STORAGE_SQLITE:
+        return
     os.makedirs(os.path.dirname(path), exist_ok=True)
     tmp_path = f"{path}.tmp"
     with open(tmp_path, "w", encoding="utf-8") as fh: