Fix stale multi-model DEB inputs

src/analysis/trend_engine.py

@@ -20,6 +20,7 @@ from src.analysis.deb_hourly_consensus import build_deb_hourly_consensus_path
 from src.analysis.settlement_rounding import apply_city_settlement, is_exact_settlement_city
 from src.data_collection.city_registry import CITY_REGISTRY
 from src.data_collection.city_risk_profiles import get_city_risk_profile
+from src.data_collection.multi_model_freshness import multi_model_forecasts_for_local_date
 
 SETTLEMENT_SOURCE_LABELS = {
     "metar": "METAR",
@@ -464,11 +465,6 @@ def analyze_weather_trend(
     if weather_data.get("cwa_forecast") is not None:
         current_forecasts["CWA(台气象)"] = _sf(weather_data.get("cwa_forecast"))
 
-    mm_forecasts = weather_data.get("multi_model", {}).get("forecasts", {})
-    for m_name, m_val in mm_forecasts.items():
-        if m_val is not None and not _is_excluded_model_name(m_name):
-            current_forecasts[m_name] = _sf(m_val)
-
     forecast_highs = [h for h in current_forecasts.values() if h is not None]
     forecast_high = max(forecast_highs) if forecast_highs else None
     forecast_median = (
@@ -545,11 +541,18 @@ def analyze_weather_trend(
                 current_forecasts["Open-Meteo"] = local_day_high
         except Exception:
             pass
-        forecast_highs = [h for h in current_forecasts.values() if h is not None]
-        forecast_high = max(forecast_highs) if forecast_highs else None
-        forecast_median = (
-            sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None
-        )
+    mm_forecasts = multi_model_forecasts_for_local_date(
+        weather_data.get("multi_model", {}),
+        local_date_str,
+    )
+    for m_name, m_val in mm_forecasts.items():
+        if m_val is not None and not _is_excluded_model_name(m_name):
+            current_forecasts[m_name] = _sf(m_val)
+    forecast_highs = [h for h in current_forecasts.values() if h is not None]
+    forecast_high = max(forecast_highs) if forecast_highs else None
+    forecast_median = (
+        sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None
+    )
 
     # === DEB ===
     deb_prediction = None

src/data_collection/forecast_source_bundle.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from typing import Any, Dict
 
+from src.data_collection.multi_model_freshness import multi_model_has_current_window
+
 
 def _open_meteo_cache_key(
     lat: float,
@@ -58,8 +60,9 @@ def _read_open_meteo_bundle_from_cache(
         )
         with collector._multi_model_cache_lock:
             mm_cached = collector._multi_model_cache.get(mm_key)
-        if mm_cached and isinstance(mm_cached.get("data"), dict):
-            results["multi_model"] = dict(mm_cached["data"])
+        mm_data = mm_cached.get("data") if isinstance(mm_cached, dict) else None
+        if isinstance(mm_data, dict) and multi_model_has_current_window(mm_data):
+            results["multi_model"] = dict(mm_data)
     return results
 
 

src/data_collection/multi_model_freshness.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+from typing import Any, Dict, Optional
+
+
+def _parse_date(value: Any) -> Optional[date]:
+    text = str(value or "").strip()
+    if not text:
+        return None
+    try:
+        return datetime.fromisoformat(text[:10]).date()
+    except Exception:
+        return None
+
+
+def _numeric(value: Any) -> Optional[float]:
+    if value is None:
+        return None
+    try:
+        return float(value)
+    except Exception:
+        return None
+
+
+def _date_strings_from_hourly(multi_model: Dict[str, Any]) -> set[str]:
+    dates = set()
+    for raw_time in multi_model.get("hourly_times") or []:
+        text = str(raw_time or "")
+        if len(text) >= 10:
+            dates.add(text[:10])
+    return dates
+
+
+def _has_numeric_hourly_for_date(multi_model: Dict[str, Any], local_date: str) -> bool:
+    times = multi_model.get("hourly_times") or []
+    forecasts = multi_model.get("hourly_forecasts") or {}
+    if not isinstance(forecasts, dict):
+        return False
+    for idx, raw_time in enumerate(times):
+        if not str(raw_time or "").startswith(local_date):
+            continue
+        for values in forecasts.values():
+            if isinstance(values, list) and idx < len(values) and _numeric(values[idx]) is not None:
+                return True
+    return False
+
+
+def multi_model_has_current_window(multi_model: Any, *, today: Optional[date] = None) -> bool:
+    """Return False when cached multi-model data is definitely older than today."""
+    if not isinstance(multi_model, dict):
+        return False
+    today = today or datetime.now(timezone.utc).date()
+    raw_date_values = []
+    raw_date_values.extend(multi_model.get("dates") or [])
+    daily = multi_model.get("daily_forecasts")
+    if isinstance(daily, dict):
+        raw_date_values.extend(daily.keys())
+    raw_date_values.extend(_date_strings_from_hourly(multi_model))
+    parsed_dates = [parsed for raw in raw_date_values for parsed in [_parse_date(raw)] if parsed]
+    if not parsed_dates:
+        return True
+    return max(parsed_dates) >= today
+
+
+def multi_model_covers_local_date(multi_model: Any, local_date: str) -> bool:
+    if not isinstance(multi_model, dict):
+        return False
+    wanted = str(local_date or "").strip()
+    if not wanted:
+        return bool(multi_model.get("forecasts"))
+
+    daily = multi_model.get("daily_forecasts") if isinstance(multi_model.get("daily_forecasts"), dict) else {}
+    day_models = daily.get(wanted) if isinstance(daily.get(wanted), dict) else {}
+    if any(_numeric(value) is not None for value in day_models.values()):
+        return True
+
+    if _has_numeric_hourly_for_date(multi_model, wanted):
+        return True
+
+    dates = [str(value) for value in (multi_model.get("dates") or [])]
+    if dates:
+        return wanted in dates
+
+    has_dated_payload = bool(daily) or bool(multi_model.get("hourly_times"))
+    return bool(multi_model.get("forecasts")) and not has_dated_payload
+
+
+def multi_model_forecasts_for_local_date(multi_model: Any, local_date: str) -> Dict[str, float]:
+    if not isinstance(multi_model, dict) or not multi_model_covers_local_date(multi_model, local_date):
+        return {}
+    wanted = str(local_date or "").strip()
+    models: Dict[str, float] = {}
+
+    daily = multi_model.get("daily_forecasts") if isinstance(multi_model.get("daily_forecasts"), dict) else {}
+    day_models = daily.get(wanted) if isinstance(daily.get(wanted), dict) else {}
+    for model, value in day_models.items():
+        parsed = _numeric(value)
+        if parsed is not None:
+            models[str(model)] = parsed
+
+    times = multi_model.get("hourly_times") or []
+    hourly = multi_model.get("hourly_forecasts") if isinstance(multi_model.get("hourly_forecasts"), dict) else {}
+    day_indexes = [
+        idx
+        for idx, raw_time in enumerate(times)
+        if wanted and str(raw_time or "").startswith(wanted)
+    ]
+    for model, values in hourly.items():
+        if not isinstance(values, list):
+            continue
+        day_values = [
+            parsed
+            for idx in day_indexes
+            if idx < len(values)
+            for parsed in [_numeric(values[idx])]
+            if parsed is not None
+        ]
+        if day_values:
+            current = models.get(str(model))
+            models[str(model)] = max(day_values) if current is None else max(current, max(day_values))
+
+    if models:
+        return models
+
+    forecasts = multi_model.get("forecasts") if isinstance(multi_model.get("forecasts"), dict) else {}
+    for model, value in forecasts.items():
+        parsed = _numeric(value)
+        if parsed is not None:
+            models[str(model)] = parsed
+    return models

src/data_collection/nws_open_meteo_sources.py

@@ -6,6 +6,7 @@ from typing import Any, Dict, Optional
 
 from loguru import logger
 
+from src.data_collection.multi_model_freshness import multi_model_has_current_window
 from src.utils.metrics import record_source_call
 
 
@@ -868,15 +869,17 @@ class NwsOpenMeteoSourceMixin:
                 logger.debug(f"Open-Meteo Multi-model 冷却期中，跳过请求，还需 {remaining}s")
                 with self._multi_model_cache_lock:
                     stale = self._multi_model_cache.get(cache_key)
-                    if stale and isinstance(stale.get("data"), dict):
+                    stale_data = stale.get("data") if isinstance(stale, dict) else None
+                    if isinstance(stale_data, dict) and multi_model_has_current_window(stale_data):
                         record_source_call("open_meteo", "multi_model", "stale_cache", (time.perf_counter() - started) * 1000.0)
-                        return dict(stale["data"])
+                        return dict(stale_data)
                 self._load_open_meteo_disk_cache()
                 with self._multi_model_cache_lock:
                     stale2 = self._multi_model_cache.get(cache_key)
-                    if stale2 and isinstance(stale2.get("data"), dict):
+                    stale2_data = stale2.get("data") if isinstance(stale2, dict) else None
+                    if isinstance(stale2_data, dict) and multi_model_has_current_window(stale2_data):
                         record_source_call("open_meteo", "multi_model", "disk_fallback", (time.perf_counter() - started) * 1000.0)
-                        return dict(stale2["data"])
+                        return dict(stale2_data)
                 record_source_call("open_meteo", "multi_model", "cooldown_skip", (time.perf_counter() - started) * 1000.0)
                 return None
 
@@ -888,6 +891,11 @@ class NwsOpenMeteoSourceMixin:
                 < self.open_meteo_multi_model_cache_ttl_sec
             ):
                 cached_data = cached.get("data")
+                if isinstance(cached_data, dict):
+                    if not multi_model_has_current_window(cached_data):
+                        self._multi_model_cache.pop(cache_key, None)
+                        record_source_call("open_meteo", "multi_model", "expired_cache_skip", (time.perf_counter() - started) * 1000.0)
+                        cached_data = None
                 if isinstance(cached_data, dict):
                     record_source_call("open_meteo", "multi_model", "cache_hit", (time.perf_counter() - started) * 1000.0)
                     return dict(cached_data)
@@ -996,8 +1004,9 @@ class NwsOpenMeteoSourceMixin:
                 logger.warning(f"Multi-model API 请求失败: {e}")
             with self._multi_model_cache_lock:
                 stale = self._multi_model_cache.get(cache_key)
-                if stale and isinstance(stale.get("data"), dict):
-                    fallback = dict(stale["data"])
+                stale_data = stale.get("data") if isinstance(stale, dict) else None
+                if isinstance(stale_data, dict) and multi_model_has_current_window(stale_data):
+                    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