From f60f5601d9d2d51c2161e6e722e0a954253b39b3 Mon Sep 17 00:00:00 2001
From: "2569718930@qq.com" <2569718930@qq.com>
Date: Fri, 27 Feb 2026 01:43:17 +0800
Subject: [PATCH] feat(ai): integrate Dynamic Ensemble Blending & simplify user
 output

---
 bot_listener.py                  |  48 ++++++++----
 src/analysis/ai_analyzer.py      |   2 +-
 src/analysis/deb_algorithm.py    | 123 +++++++++++++++++++++++++++++++
 src/data_mining/fetch_history.py |  85 +++++++++++++++++++++
 4 files changed, 242 insertions(+), 16 deletions(-)
 create mode 100644 src/analysis/deb_algorithm.py
 create mode 100644 src/data_mining/fetch_history.py

diff --git a/bot_listener.py b/bot_listener.py
index 1324200e..e422eeb4 100644
--- a/bot_listener.py
+++ b/bot_listener.py
@@ -13,8 +13,9 @@ if project_root not in sys.path:
 from src.utils.config_loader import load_config  # type: ignore
 from src.data_collection.weather_sources import WeatherDataCollector  # type: ignore
 from src.data_collection.city_risk_profiles import get_city_risk_profile, format_risk_warning  # type: ignore
+from src.analysis.deb_algorithm import calculate_dynamic_weights, update_daily_record
 
-def analyze_weather_trend(weather_data, temp_symbol):
+def analyze_weather_trend(weather_data, temp_symbol, city_name=None):
     '''根据实测与预测分析气温态势，增加峰值时刻预测'''
     insights: List[str] = []
     ai_features: List[str] = []
@@ -27,6 +28,7 @@ def analyze_weather_trend(weather_data, temp_symbol):
     
     if not metar or not open_meteo:
         return "", ""
+
         
     curr_temp = metar.get("current", {}).get("temp")
     max_so_far = metar.get("current", {}).get("max_temp_so_far")  # 今日实测最高
@@ -35,17 +37,23 @@ def analyze_weather_trend(weather_data, temp_symbol):
     times = hourly.get("time", [])
     temps = hourly.get("temperature_2m", [])
     
-    # === 核心：整合多源预报最高温 ===
-    forecast_highs = [daily.get("temperature_2m_max", [None])[0]]
+    # === 新核心：动态集合权重预报 (DEB) ===
+    # 抽取各个确定性预报值构成的字典
+    current_forecasts = {}
+    if daily.get("temperature_2m_max"):
+        current_forecasts["Open-Meteo"] = daily.get("temperature_2m_max")[0]
     if mb.get("today_high") is not None:
-        forecast_highs.append(mb["today_high"])
+        current_forecasts["Meteoblue"] = mb.get("today_high")
     if nws.get("today_high") is not None:
-        forecast_highs.append(nws["today_high"])
-    for mv in weather_data.get("multi_model", {}).get("forecasts", {}).values():
-        if mv is not None:
-            forecast_highs.append(mv)
-    
-    forecast_highs = [h for h in forecast_highs if h is not None]
+        current_forecasts["NWS"] = nws.get("today_high")
+        
+    mm_forecasts = weather_data.get("multi_model", {}).get("forecasts", {})
+    for m_name, m_val in mm_forecasts.items():
+        if m_val is not None:
+            current_forecasts[m_name] = m_val
+            
+    # 从 URL/入参里我们暂时拿不到城名，为了 DEB 追溯我们在后方的总控那里提取。这里的 analyze_weather_trend 主要计算最高预留。
+    forecast_highs = [h for h in current_forecasts.values() if h is not None]
     forecast_high = max(forecast_highs) if forecast_highs else None
     min_forecast_high = min(forecast_highs) if forecast_highs else forecast_high
     forecast_median = sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None
@@ -62,6 +70,19 @@ def analyze_weather_trend(weather_data, temp_symbol):
         local_date_str = datetime.now().strftime("%Y-%m-%d")
         local_hour = datetime.now().hour
 
+    # === DEB 融合渲染 ===
+    if city_name and current_forecasts:
+        blended_high, weight_info = calculate_dynamic_weights(city_name, current_forecasts)
+        if blended_high is not None:
+            insights.insert(0, f"🧬 <b>DEB 融合预测</b>：<b>{blended_high}{temp_symbol}</b> ({weight_info})")
+            ai_features.append(f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。")
+            
+        # 顺便把今天的预测记录下来供之后回测用
+        try:
+            update_daily_record(city_name, local_date_str, current_forecasts, max_so_far)
+        except:
+            pass
+
     # === METAR 趋势分析 (移到前部判断降温) ===
     recent_temps = metar.get("recent_temps", [])
     trend_desc = ""
@@ -160,16 +181,13 @@ def analyze_weather_trend(weather_data, temp_symbol):
         if max_so_far > forecast_high + 0.5:
             exceed_by = max_so_far - forecast_high
             bt_msg = f"🚨 <b>实测已超预报</b>：{max_so_far}{temp_symbol} 超过上限 {forecast_high}{temp_symbol}（+{exceed_by:.1f}°）。"
-            if trend_desc: bt_msg += f"\n{trend_desc}"
             insights.append(bt_msg)
             ai_features.append(f"🚨 异常: 实测已冲破所有预报上限 ({max_so_far}{temp_symbol} vs {forecast_high}{temp_symbol})。")
-            ai_features.append(trend_desc)
+            if trend_desc: ai_features.append(trend_desc)
         else:
             if trend_desc:
-                insights.append(trend_desc)
                 ai_features.append(trend_desc)
     elif trend_desc:
-        insights.append(trend_desc)
         ai_features.append(trend_desc)
 
     # === 结算取整分析 ===
@@ -588,7 +606,7 @@ def start_bot():
                     msg_lines.append(f"   {prefix} {cloud_desc} | 👁️ {vis or 10}mi | 💨 {wind or 0}kt")
 
             # --- 5. 态势特征提取 ---
-            feature_str, ai_context = analyze_weather_trend(weather_data, temp_symbol)
+            feature_str, ai_context = analyze_weather_trend(weather_data, temp_symbol, city_name)
             if feature_str:
                 # 仅将最核心的信息展示给用户作为"态势分析"
                 # 但后面会把更全的数据传给 AI
diff --git a/src/analysis/ai_analyzer.py b/src/analysis/ai_analyzer.py
index d96777a8..3c5074f7 100644
--- a/src/analysis/ai_analyzer.py
+++ b/src/analysis/ai_analyzer.py
@@ -38,7 +38,7 @@ def get_ai_analysis(weather_insights: str, city_name: str, temp_symbol: str) ->
 2. 严格按照以下 HTML 格式输出:
 
 🤖 <b>Groq AI 决策</b>
-- 🎲 盘口: [一句话指出结算在哪里博弈，以及目前是否到了最热时段。例如：距最热时段还有3小时，目前预计在27或28之间博弈，有突破可能。]
+- 🎲 盘口: [一句话指出结算在哪里博弈，以及目前是否到了最热时段。如果已经处于明确的降温期，请直接给出死盘结论（如：已过最热点且降温，今日锁定在 X 度结算，悬念终止），不要再说“博弈”。]
 - 💡 逻辑: [不要重复模版例子！请使用一句话提炼机场实测(如风速风向、云量、气温变化趋势)及热力动力因子。例如：实测吹强劲西南风(15kt)伴随云量减少，辐射加热强劲，破预报阻力非常小。]
 - 🎯 置信度: [1-10]/10
 """
diff --git a/src/analysis/deb_algorithm.py b/src/analysis/deb_algorithm.py
new file mode 100644
index 00000000..2fa3c290
--- /dev/null
+++ b/src/analysis/deb_algorithm.py
@@ -0,0 +1,123 @@
+import os
+import json
+import logging
+from datetime import datetime, timedelta
+
+def load_history(filepath):
+    if os.path.exists(filepath):
+        try:
+            with open(filepath, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        except:
+            return {}
+    return {}
+
+def save_history(filepath, data):
+    with open(filepath, 'w', encoding='utf-8') as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+
+def update_daily_record(city_name, date_str, forecasts, actual_high):
+    """
+    保存/更新某城市某天的各个模型预报与最终实测值
+    forecasts: dict, 例如 {"ECMWF": 28.5, "GFS": 30.0, ...}
+    actual_high: float, 最终实测最高温
+    """
+    project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    history_file = os.path.join(project_root, 'data', 'daily_records.json')
+    
+    data = load_history(history_file)
+    if city_name not in data:
+        data[city_name] = {}
+        
+    if date_str not in data[city_name]:
+        data[city_name][date_str] = {}
+    
+    data[city_name][date_str]['forecasts'] = forecasts
+    # 只要仍在更新或者已经结束，都记录最新高点
+    data[city_name][date_str]['actual_high'] = actual_high
+    
+    save_history(history_file, data)
+
+def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7):
+    """
+    计算动态权重融合 (Dynamic Ensemble Blending, DEB)
+    根据过去 N 天各模型的 Mean Absolute Error (MAE) 计算倒数权重
+    返回: blended_high (融合预报值), weights_info (权重展示字符串)
+    """
+    project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    history_file = os.path.join(project_root, 'data', 'daily_records.json')
+    data = load_history(history_file)
+    
+    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)
+        return round(avg, 1), "等权平均(历史数据不足)"
+        
+    # 获取过去 lookback_days 天的有 actual_high 的记录
+    city_data = data[city_name]
+    sorted_dates = sorted(city_data.keys(), reverse=True)
+    
+    # 我们只用真正结清（或者有比较准确最高温）的历史来算误差
+    # 这边简化：凡是有 actual_high 的都算进去
+    errors = {model: [] for model in current_forecasts.keys()}
+    
+    days_used = 0
+    for date_str in sorted_dates:
+        # 跳过今天，今天还没出最终结果
+        if date_str == datetime.now().strftime("%Y-%m-%d"):
+            continue
+            
+        record = city_data[date_str]
+        actual = record.get('actual_high')
+        past_forecasts = record.get('forecasts', {})
+        
+        if actual is None:
+            continue
+            
+        for model in current_forecasts.keys():
+            if model in past_forecasts and past_forecasts[model] is not None:
+                errors[model].append(abs(past_forecasts[model] - actual))
+                
+        days_used += 1
+        if days_used >= lookback_days:
+            break
+            
+    # 如果有效历史天数 < 2 天，还是使用等权
+    if days_used < 2:
+        valid_vals = [v for v in current_forecasts.values() if v is not None]
+        avg = sum(valid_vals) / len(valid_vals)
+        return round(avg, 1), f"等权平均(由于仅{days_used}天历史)"
+        
+    # 计算 MAE
+    maes = {}
+    for model, err_list in errors.items():
+        if err_list:
+            maes[model] = sum(err_list) / len(err_list)
+        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}
+    
+    total_inv = sum(inverse_errors.values())
+    if total_inv == 0:
+        return None, "权重计算异常"
+        
+    weights = {m: inv / total_inv for m, inv in inverse_errors.items()}
+    
+    # 计算加权最高温
+    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]:
+         weight_str_parts.append(f"{m}({w*100:.0f}%,MAE:{maes[m]:.1f}°)")
+         
+    return round(blended_high, 1), " | ".join(weight_str_parts)
+
diff --git a/src/data_mining/fetch_history.py b/src/data_mining/fetch_history.py
new file mode 100644
index 00000000..64e7f0b8
--- /dev/null
+++ b/src/data_mining/fetch_history.py
@@ -0,0 +1,85 @@
+import os
+import sys
+import json
+import logging
+from datetime import datetime
+import pandas as pd
+import requests
+
+# Set up logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+def fetch_historical_data_for_city(city_info, output_dir):
+    city_name = city_info['city']
+    lat = city_info['latitude']
+    lon = city_info['longitude']
+    
+    # We will fetch data from Jan 1, 2023 to yesterday (or to latest available)
+    start_date = "2023-01-01"
+    end_date = "2025-12-31"  # API handles dates in the future up to latest available archive usually 
+    # For safety let's use a dynamic yesterday end_date
+    import datetime
+    today = datetime.datetime.now()
+    yesterday = (today - datetime.timedelta(days=2)).strftime("%Y-%m-%d")
+    
+    url = (
+        f"https://archive-api.open-meteo.com/v1/archive?latitude={lat}&longitude={lon}"
+        f"&start_date={start_date}&end_date={yesterday}"
+        "&hourly=temperature_2m,relative_humidity_2m,wind_speed_10m,wind_direction_10m,"
+        "cloud_cover,shortwave_radiation,precipitation,surface_pressure"
+        "&timezone=auto"
+    )
+    
+    logging.info(f"Downloading historical data for {city_name} (Lat: {lat}, Lon: {lon})...")
+    
+    try:
+        response = requests.get(url, timeout=60)
+        response.raise_for_status()
+        data = response.json()
+        
+        if "hourly" not in data:
+            logging.error(f"Failed to find 'hourly' data for {city_name}.")
+            return
+            
+        hourly_data = data["hourly"]
+        
+        # Convert to pandas DataFrame
+        df = pd.DataFrame(hourly_data)
+        
+        # Save to CSV
+        output_path = os.path.join(output_dir, f"{city_name.replace(' ', '_').lower()}_historical.csv")
+        df.to_csv(output_path, index=False)
+        
+        logging.info(f"✅ Successfully saved historical data for {city_name} to {output_path}. Shape: {df.shape}")
+        
+    except requests.exceptions.RequestException as e:
+        logging.error(f"❌ Network error while fetching data for {city_name}: {e}")
+    except Exception as e:
+        logging.error(f"❌ Error processing data for {city_name}: {e}")
+
+def main():
+    project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    config_path = os.path.join(project_root, 'config', 'config.yaml')
+    output_dir = os.path.join(project_root, 'data', 'historical')
+    
+    os.makedirs(output_dir, exist_ok=True)
+    
+    # Load config
+    try:
+        import yaml
+        with open(config_path, 'r', encoding='utf-8') as f:
+            config = yaml.safe_load(f)
+    except Exception as e:
+        logging.error(f"Failed to load {config_path}: {e}")
+        sys.exit(1)
+        
+    cities = config.get('cities', [])
+    if not cities:
+        logging.warning("No cities found in config.yaml")
+        return
+        
+    for city_info in cities:
+        fetch_historical_data_for_city(city_info, output_dir)
+        
+if __name__ == "__main__":
+    main()