feat(ai): integrate Dynamic Ensemble Blending & simplify user output
This commit is contained in:
2569718930@qq.com
+33
-15
@@ -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
|
||||
|
||||
@@ -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
|
||||
"""
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user