feat: build DEB hourly consensus for peak windows

src/analysis/deb_algorithm.py

@@ -1092,6 +1092,27 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
 
     返回: blended_high (融合预报值), weights_info (权重展示字符串)
     """
+    components = calculate_dynamic_weight_components(
+        city_name,
+        current_forecasts,
+        lookback_days=lookback_days,
+        decay_factor=decay_factor,
+    )
+    forecasts = components.get("forecasts") or {}
+    weights = components.get("weights") or {}
+    if not forecasts or not weights:
+        return components.get("prediction"), components.get("weights_info") or "暂无模型数据"
+    blended_high = sum(forecasts[m] * weights[m] for m in weights if m in forecasts)
+    return round(blended_high, 1), components.get("weights_info") or "权重计算异常"
+
+
+def calculate_dynamic_weight_components(
+    city_name,
+    current_forecasts,
+    lookback_days=7,
+    decay_factor=0.85,
+):
+    """Return DEB forecast representatives and model weights for reuse by hourly paths."""
     project_root = os.path.dirname(
         os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     )
@@ -1105,23 +1126,41 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
             if v is not None and not _is_excluded_model_name(k)
         ]
     )
-    current_forecasts = _collapse_forecasts_for_deb(current_forecasts)
-    dedup_note = "家族去重" if raw_forecast_count > len(current_forecasts) else ""
+    forecasts = _collapse_forecasts_for_deb(current_forecasts)
+    dedup_note = "家族去重" if raw_forecast_count > len(forecasts) else ""
+    valid_vals = [v for v in forecasts.values() if v is not None]
+    if not valid_vals:
+        return {
+            "prediction": None,
+            "weights": {},
+            "forecasts": {},
+            "maes": {},
+            "weights_info": "暂无模型数据",
+            "days_used": 0,
+            "dedup_note": dedup_note,
+        }
+
+    def _equal_weight_result(note: str, days_used: int):
+        weights = {model: 1.0 / len(forecasts) for model in forecasts}
+        weights_info = f"{note} | {dedup_note}" if dedup_note else note
+        prediction = sum(forecasts[m] * weights[m] for m in weights)
+        return {
+            "prediction": round(prediction, 1),
+            "weights": weights,
+            "forecasts": forecasts,
+            "maes": {},
+            "weights_info": weights_info,
+            "days_used": days_used,
+            "dedup_note": dedup_note,
+        }
 
     if city_name not in data or not data[city_name]:
-        valid_vals = [v for v in current_forecasts.values() if v is not None]
-        if not valid_vals:
-            return None, "暂无模型数据"
-        avg = sum(valid_vals) / len(valid_vals)
-        note = "等权平均(历史数据不足)"
-        if dedup_note:
-            note = f"{note} | {dedup_note}"
-        return round(avg, 1), note
+        return _equal_weight_result("等权平均(历史数据不足)", 0)
 
     city_data = data[city_name]
     sorted_dates = sorted(city_data.keys(), reverse=True)
 
-    errors: dict = {model: [] for model in current_forecasts.keys()}
+    errors: dict = {model: [] for model in forecasts.keys()}
     days_used = 0
     for date_str in sorted_dates:
         if date_str == datetime.now().strftime("%Y-%m-%d"):
@@ -1137,7 +1176,7 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
 
         decay_weight = decay_factor ** days_used
 
-        for model in current_forecasts.keys():
+        for model in forecasts.keys():
             if model in past_forecasts and past_forecasts[model] is not None:
                 try:
                     pv = float(past_forecasts[model])
@@ -1145,7 +1184,6 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
                 except (TypeError, ValueError):
                     continue
                 daily_error = abs(pv - av)
-                # Blend with hourly error when available
                 h_err = (
                     past_hourly_error.get(model)
                     if isinstance(past_hourly_error, dict)
@@ -1159,16 +1197,8 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
             break
 
     if days_used < 2:
-        valid_vals = [v for v in current_forecasts.values() if v is not None]
-        if not valid_vals:
-            return None, f"暂无有效模型数据(由于仅{days_used}天历史)"
-        avg = sum(valid_vals) / len(valid_vals)
-        note = f"等权平均(由于仅{days_used}天历史)"
-        if dedup_note:
-            note = f"{note} | {dedup_note}"
-        return round(avg, 1), note
+        return _equal_weight_result(f"等权平均(由于仅{days_used}天历史)", days_used)
 
-    # 计算加权 MAE（时间衰减）
     maes = {}
     for model, err_weighted in errors.items():
         if err_weighted:
@@ -1180,25 +1210,27 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
         else:
             maes[model] = 2.0
 
-    # 计算权重（用 MAE 的倒数，误差越小权重越大；加 0.1 防止除以0）
     inverse_errors = {
         m: 1.0 / (mae + 0.1)
         for m, mae in maes.items()
-        if current_forecasts.get(m) is not None
+        if forecasts.get(m) is not None
     }
 
     total_inv = sum(inverse_errors.values())
     if total_inv == 0:
-        return None, "权重计算异常"
+        return {
+            "prediction": None,
+            "weights": {},
+            "forecasts": forecasts,
+            "maes": maes,
+            "weights_info": "权重计算异常",
+            "days_used": days_used,
+            "dedup_note": dedup_note,
+        }
 
     weights = {m: inv / total_inv for m, inv in inverse_errors.items()}
+    blended_high = sum(forecasts[m] * weights[m] for m in weights)
 
-    # 计算加权最高温
-    blended_high = 0.0
-    for m in weights.keys():
-        blended_high += current_forecasts[m] * weights[m]
-
-    # 格式化权重信息，挑选前权重最高的2-3个模型展示
     sorted_models = sorted(weights.items(), key=lambda x: x[1], reverse=True)
     weight_str_parts = []
     for m, w in sorted_models[:3]:
@@ -1206,7 +1238,15 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7, dec
     if dedup_note:
         weight_str_parts.append(dedup_note)
 
-    return round(blended_high, 1), " | ".join(weight_str_parts)
+    return {
+        "prediction": round(blended_high, 1),
+        "weights": weights,
+        "forecasts": forecasts,
+        "maes": maes,
+        "weights_info": " | ".join(weight_str_parts),
+        "days_used": days_used,
+        "dedup_note": dedup_note,
+    }
 
 
 def calculate_deb_prediction(

src/analysis/deb_hourly_consensus.py

@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from src.analysis.deb_algorithm import calculate_dynamic_weight_components
+
+DEB_HOURLY_CONSENSUS_VERSION = "deb_hourly_consensus.v1"
+
+
+def _to_float(value: Any) -> Optional[float]:
+    try:
+        result = float(value)
+    except (TypeError, ValueError):
+        return None
+    if result != result:
+        return None
+    return result
+
+
+def _time_part(value: Any) -> str:
+    text = str(value or "").strip()
+    if "T" in text:
+        text = text.split("T", 1)[1]
+    if " " in text:
+        text = text.rsplit(" ", 1)[-1]
+    return text[:5]
+
+
+def _matches_local_date(value: Any, local_date: Optional[str]) -> bool:
+    if not local_date:
+        return True
+    text = str(value or "").strip()
+    if "T" not in text and " " not in text:
+        return True
+    return text.startswith(local_date)
+
+
+def _weighted_value_at_index(
+    index: int,
+    hourly_forecasts: Dict[str, Any],
+    weights: Dict[str, float],
+) -> Optional[float]:
+    weighted_sum = 0.0
+    weight_sum = 0.0
+    for model_name, model_weight in weights.items():
+        series = hourly_forecasts.get(model_name)
+        if not isinstance(series, (list, tuple)) or index >= len(series):
+            continue
+        value = _to_float(series[index])
+        if value is None:
+            continue
+        weighted_sum += value * model_weight
+        weight_sum += model_weight
+    if weight_sum <= 0:
+        return None
+    return weighted_sum / weight_sum
+
+
+def build_deb_hourly_consensus_path(
+    *,
+    city: str,
+    hourly_times: List[Any],
+    hourly_forecasts: Dict[str, Any],
+    daily_forecasts: Dict[str, Any],
+    deb_prediction: Optional[float],
+    local_date: Optional[str] = None,
+) -> Optional[Dict[str, Any]]:
+    if not hourly_times or not isinstance(hourly_forecasts, dict):
+        return None
+
+    components = calculate_dynamic_weight_components(city, daily_forecasts or {})
+    weights = {
+        model: float(weight)
+        for model, weight in (components.get("weights") or {}).items()
+        if model in hourly_forecasts and _to_float(weight) is not None
+    }
+    if not weights:
+        return None
+
+    times: List[str] = []
+    raw_temps: List[Optional[float]] = []
+    for idx, raw_time in enumerate(hourly_times):
+        if not _matches_local_date(raw_time, local_date):
+            continue
+        value = _weighted_value_at_index(idx, hourly_forecasts, weights)
+        if value is None:
+            continue
+        times.append(_time_part(raw_time))
+        raw_temps.append(round(value, 3))
+
+    numeric_raw = [value for value in raw_temps if value is not None]
+    if not times or not numeric_raw:
+        return None
+
+    deb_value = _to_float(deb_prediction)
+    anchor_adjustment = 0.0
+    if deb_value is not None:
+        anchor_adjustment = deb_value - max(numeric_raw)
+    temps = [
+        round(value + anchor_adjustment, 1) if value is not None else None
+        for value in raw_temps
+    ]
+
+    return {
+        "version": DEB_HOURLY_CONSENSUS_VERSION,
+        "source": DEB_HOURLY_CONSENSUS_VERSION,
+        "base_source": "multi_model_hourly_deb_weights",
+        "times": times,
+        "temps": temps,
+        "raw_temps": [round(value, 1) if value is not None else None for value in raw_temps],
+        "weights": {model: round(weight, 4) for model, weight in weights.items()},
+        "weights_info": components.get("weights_info") or "",
+        "anchor_adjustment": round(anchor_adjustment, 3),
+    }

src/analysis/deb_hourly_correction.py

@@ -280,6 +280,7 @@ def build_deb_hourly_path(
     peak_first_h: Optional[int],
     peak_last_h: Optional[int],
     corrector: HourlyPeakCorrector,
+    base_source: str = "hourly_plus_deb_offset",
 ) -> Dict[str, Any]:
     deb_value = _to_float(deb_prediction)
     numeric_base = [_to_float(value) for value in hourly_temps]
@@ -305,7 +306,7 @@ def build_deb_hourly_path(
         "version": DEB_HOURLY_PEAK_CORRECTED_VERSION,
         "times": applied["times"],
         "temps": applied["temps"],
-        "base_source": "hourly_plus_deb_offset",
+        "base_source": base_source,
         "base_offset": round(offset, 3),
         "correction": {
             "version": applied["version"],

src/analysis/trend_engine.py

@@ -16,6 +16,7 @@ from src.analysis.deb_algorithm import (
     update_daily_record,
     _is_excluded_model_name,
 )
+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
@@ -81,6 +82,29 @@ def _resolve_peak_hours(
     open_meteo_peak: Optional[Any] = None,
 ) -> List[str]:
     """Resolve the local high-temperature window, preferring multi-model hourly consensus."""
+    deb = weather_data.get("deb") if isinstance(weather_data, dict) else {}
+    if isinstance(deb, dict):
+        consensus = deb.get("hourly_consensus")
+        if isinstance(consensus, dict):
+            c_times = consensus.get("times") or []
+            c_temps = consensus.get("temps") or []
+            hourly_values: List[Tuple[str, float]] = []
+            for raw_time, raw_temp in zip(c_times, c_temps):
+                t_str = str(raw_time or "")
+                if "T" in t_str and not t_str.startswith(local_date_str):
+                    continue
+                time_part = t_str.split("T", 1)[1][:5] if "T" in t_str else t_str[:5]
+                try:
+                    hour = int(time_part[:2])
+                except Exception:
+                    continue
+                value = _sf(raw_temp)
+                if value is not None and 8 <= hour <= 19:
+                    hourly_values.append((time_part, value))
+            peak_hours = _peak_hours_from_hourly_values(hourly_values)
+            if peak_hours:
+                return peak_hours
+
     multi_model = weather_data.get("multi_model") if isinstance(weather_data, dict) else {}
     if isinstance(multi_model, dict):
         hourly_times = multi_model.get("hourly_times") or []
@@ -420,6 +444,7 @@ def analyze_weather_trend(
     # === DEB ===
     deb_prediction = None
     deb_raw_prediction = None
+    deb_hourly_consensus = None
     deb_version = None
     deb_bias_adjustment = 0.0
     deb_bias_samples = 0
@@ -506,10 +531,30 @@ def analyze_weather_trend(
     is_cooling = trend_direction == "falling"
 
     om_today = _sf(current_forecasts.get("Open-Meteo"))
+    if city_name and deb_prediction is not None:
+        mm = weather_data.get("multi_model") or {}
+        if isinstance(mm, dict):
+            deb_hourly_consensus = build_deb_hourly_consensus_path(
+                city=city_name,
+                hourly_times=mm.get("hourly_times") or [],
+                hourly_forecasts=mm.get("hourly_forecasts") or {},
+                daily_forecasts=current_forecasts,
+                deb_prediction=deb_prediction,
+                local_date=local_date_str,
+            )
 
     # === Peak hours ===
+    peak_weather_data = weather_data
+    if deb_hourly_consensus:
+        peak_weather_data = {
+            **weather_data,
+            "deb": {
+                **(weather_data.get("deb") or {}),
+                "hourly_consensus": deb_hourly_consensus,
+            },
+        }
     peak_hours = _resolve_peak_hours(
-        weather_data,
+        peak_weather_data,
         local_date_str,
         times,
         temps,
@@ -942,6 +987,7 @@ def analyze_weather_trend(
         "peak_hours": peak_hours,
         "deb_prediction": deb_prediction,
         "deb_raw_prediction": deb_raw_prediction,
+        "deb_hourly_consensus": deb_hourly_consensus,
         "deb_version": deb_version,
         "deb_bias_adjustment": deb_bias_adjustment,
         "deb_bias_samples": deb_bias_samples,