From 1060945d085132f269a92600bd8c22c9c1f1498f Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Thu, 12 Mar 2026 02:32:10 +0800 Subject: [PATCH] feat: Implement Dynamic Ensemble Blending (DEB) algorithm with historical data management, dynamic weight calculation, and accuracy tracking. --- bot_listener.py | 10 ++++-- src/analysis/deb_algorithm.py | 23 +++++++++++++ src/analysis/trend_engine.py | 64 ++++++++++++++++++++++++++++------- web/app.py | 24 +++++++++++-- 4 files changed, 104 insertions(+), 17 deletions(-) diff --git a/bot_listener.py b/bot_listener.py index 65976cb2..e4b0ea05 100644 --- a/bot_listener.py +++ b/bot_listener.py @@ -132,7 +132,7 @@ def start_bot(): from datetime import datetime as _dt, timedelta as _td import os as _os - from src.analysis.deb_algorithm import load_history + from src.analysis.deb_algorithm import load_history, _is_excluded_model_name from src.data_collection.city_registry import ALIASES city_input = parts[1].strip().lower() @@ -196,7 +196,11 @@ def start_bot(): continue if deb_pred is None and forecasts: - valid_preds = [float(v) for v in forecasts.values() if v is not None] + valid_preds = [ + float(v) + for k, v in forecasts.items() + if v is not None and not _is_excluded_model_name(k) + ] if valid_preds: deb_pred = round(sum(valid_preds) / len(valid_preds), 1) @@ -232,6 +236,8 @@ def start_bot(): if date_str != today_str and actual is not None: for model, pred in forecasts.items(): + if _is_excluded_model_name(model): + continue if pred is None: continue try: diff --git a/src/analysis/deb_algorithm.py b/src/analysis/deb_algorithm.py index 4994d0f4..7018aa26 100644 --- a/src/analysis/deb_algorithm.py +++ b/src/analysis/deb_algorithm.py @@ -37,6 +37,11 @@ _history_cache = {} _history_mtime = 0 +def _is_excluded_model_name(model_name: str) -> bool: + normalized = str(model_name or "").strip().lower().replace(" ", "").replace("_", "").replace("-", "") + return "meteoblue" in normalized + + def load_history(filepath): global _history_cache, _history_mtime if not os.path.exists(filepath): @@ -100,6 +105,11 @@ def update_daily_record( if date_str not in data[city_name]: data[city_name][date_str] = {} + # 统一过滤已弃用模型，避免历史/展示残留 + forecasts = { + k: v for k, v in (forecasts or {}).items() if not _is_excluded_model_name(k) + } + compact_probs = None if probabilities is not None: # Store compact: [{"v": 25, "p": 0.8}, ...] @@ -124,6 +134,13 @@ def update_daily_record( ): return + # actual_high 应该是日内最高温，理论上不应下降；防止异常写入覆盖已确认高值 + if old_actual is not None and actual_high is not None: + try: + actual_high = max(float(old_actual), float(actual_high)) + except Exception: + pass + existing["forecasts"] = forecasts existing["actual_high"] = actual_high if deb_prediction is not None: @@ -155,6 +172,12 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7): history_file = os.path.join(project_root, "data", "daily_records.json") data = load_history(history_file) + current_forecasts = { + k: v + for k, v in (current_forecasts or {}).items() + if not _is_excluded_model_name(k) + } + 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] diff --git a/src/analysis/trend_engine.py b/src/analysis/trend_engine.py index 827592b9..7d84dff6 100644 --- a/src/analysis/trend_engine.py +++ b/src/analysis/trend_engine.py @@ -6,12 +6,14 @@ for both Telegram bot and web dashboard. """ import math +from datetime import datetime, timezone, timedelta from typing import List, Optional, Tuple, Dict, Any from src.analysis.deb_algorithm import ( calculate_dynamic_weights, get_deb_accuracy, update_daily_record, + _is_excluded_model_name, ) from src.analysis.settlement_rounding import wu_round from src.data_collection.city_risk_profiles import get_city_risk_profile @@ -89,7 +91,7 @@ def analyze_weather_trend( mm_forecasts = weather_data.get("multi_model", {}).get("forecasts", {}) for m_name, m_val in mm_forecasts.items(): - if m_val is not None: + 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] @@ -100,18 +102,54 @@ def analyze_weather_trend( wind_speed = metar.get("current", {}).get("wind_speed_kt", 0) - # === Local time === - local_time_full = open_meteo.get("current", {}).get("local_time", "") - try: - local_date_str = local_time_full.split(" ")[0] - time_parts = local_time_full.split(" ")[1].split(":") - local_hour = int(time_parts[0]) - local_minute = int(time_parts[1]) if len(time_parts) > 1 else 0 - except Exception: - from datetime import datetime - local_date_str = datetime.now().strftime("%Y-%m-%d") - local_hour = datetime.now().hour - local_minute = datetime.now().minute + # === Local time/date (do not trust cached Open-Meteo local_time for date key) === + utc_offset = _sf(open_meteo.get("utc_offset")) + if utc_offset is None and city_name: + try: + from src.data_collection.city_registry import CITY_REGISTRY + + city_meta = CITY_REGISTRY.get(str(city_name).lower()) + if isinstance(city_meta, dict): + utc_offset = _sf(city_meta.get("tz_offset")) + except Exception: + pass + + city_now = None + if utc_offset is not None: + try: + city_now = datetime.now(timezone.utc).astimezone( + timezone(timedelta(seconds=int(utc_offset))) + ) + except Exception: + city_now = None + + local_time_full = str((open_meteo.get("current") or {}).get("local_time") or "").strip() + if city_now is not None: + local_date_str = city_now.strftime("%Y-%m-%d") + local_hour = city_now.hour + local_minute = city_now.minute + else: + try: + local_date_str = local_time_full.split(" ")[0] + time_parts = local_time_full.split(" ")[1].split(":") + local_hour = int(time_parts[0]) + local_minute = int(time_parts[1]) if len(time_parts) > 1 else 0 + except Exception: + fallback_now = datetime.now() + local_date_str = fallback_now.strftime("%Y-%m-%d") + local_hour = fallback_now.hour + local_minute = fallback_now.minute + + # Use METAR observation date in city local time when available (reliable for actual_high date key). + metar_obs_time_raw = str(metar.get("observation_time") or "").strip() + if metar_obs_time_raw and utc_offset is not None: + try: + metar_obs_dt = datetime.fromisoformat(metar_obs_time_raw.replace("Z", "+00:00")) + local_date_str = metar_obs_dt.astimezone( + timezone(timedelta(seconds=int(utc_offset))) + ).strftime("%Y-%m-%d") + except Exception: + pass local_hour_frac = local_hour + local_minute / 60 # === DEB === diff --git a/web/app.py b/web/app.py index 3978f317..f3cff4a7 100644 --- a/web/app.py +++ b/web/app.py @@ -126,6 +126,18 @@ def _sf(v) -> Optional[float]: return None +def _is_excluded_model_name(model_name: str) -> bool: + normalized = ( + str(model_name or "") + .strip() + .lower() + .replace(" ", "") + .replace("_", "") + .replace("-", "") + ) + return "meteoblue" in normalized + + # ────────────────────────────────────────────────────────── # Core Analysis (replicates bot_listener logic → JSON) # ────────────────────────────────────────────────────────── @@ -283,7 +295,7 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: if om_today is not None: current_forecasts["Open-Meteo"] = om_today for m, v in mm.get("forecasts", {}).items(): - if v is not None: + if v is not None and not _is_excluded_model_name(m): current_forecasts[m] = _sf(v) nws_high = _sf(raw.get("nws", {}).get("today_high")) if nws_high is not None: @@ -594,6 +606,10 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: mgm_daily = mgm.get("daily_forecasts", {}) if d_str in mgm_daily: day_m["MGM"] = _sf(mgm_daily[d_str]) + + day_m = { + m: v for m, v in day_m.items() if not _is_excluded_model_name(m) + } d_val, d_winfo = None, "" d_probs = [] @@ -914,7 +930,11 @@ def _build_city_detail_payload( "mgm_hourly": (data.get("mgm") or {}).get("hourly", []), "forecast_daily": (data.get("forecast") or {}).get("daily", []), }, - "models": data.get("multi_model") or {}, + "models": { + k: v + for k, v in (data.get("multi_model") or {}).items() + if not _is_excluded_model_name(k) + }, "probabilities": data.get("probabilities") or {"mu": None, "distribution": []}, "market_scan": market_scan, "risk": data.get("risk"),