diff --git a/bot_listener.py b/bot_listener.py index 77a2e7eb..88d0979e 100644 --- a/bot_listener.py +++ b/bot_listener.py @@ -1,7 +1,6 @@ import sys import os -from datetime import datetime -from typing import List, Dict, Any, Optional +from typing import List import telebot # type: ignore from loguru import logger # type: ignore @@ -10,39 +9,43 @@ project_root = os.path.dirname(os.path.abspath(__file__)) if project_root not in sys.path: sys.path.insert(0, project_root) -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 +from src.utils.config_loader import load_config # type: ignore # noqa: E402 +from src.data_collection.weather_sources import WeatherDataCollector # type: ignore # noqa: E402 +from src.data_collection.city_risk_profiles import get_city_risk_profile # type: ignore # noqa: E402 +from src.analysis.deb_algorithm import calculate_dynamic_weights, update_daily_record # noqa: E402 + def analyze_weather_trend(weather_data, temp_symbol, city_name=None): - '''根据实测与预测分析气温态势，增加峰值时刻预测''' + """根据实测与预测分析气温态势，增加峰值时刻预测""" insights: List[str] = [] ai_features: List[str] = [] - + metar = weather_data.get("metar", {}) open_meteo = weather_data.get("open-meteo", {}) mb = weather_data.get("meteoblue", {}) nws = weather_data.get("nws", {}) - mgm = weather_data.get("mgm", {}) - + weather_data.get("mgm", {}) + if not metar or not open_meteo: return "", "" # 数值归一化：防止 JSON 反序列化后的 str 类型炸数学运算 def _sf(v): """safe float""" - if v is None: return None - try: return float(v) - except: return None - - curr_temp = _sf(metar.get("current", {}).get("temp")) + if v is None: + return None + try: + return float(v) + except Exception: + return None + + _sf(metar.get("current", {}).get("temp")) max_so_far = _sf(metar.get("current", {}).get("max_temp_so_far")) daily = open_meteo.get("daily", {}) hourly = open_meteo.get("hourly", {}) times = hourly.get("time", []) temps = hourly.get("temperature_2m", []) - + # === 新核心：动态集合权重预报 (DEB) === # 抽取各个确定性预报值构成的字典 current_forecasts = {} @@ -52,20 +55,22 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): current_forecasts["Meteoblue"] = _sf(mb.get("today_high")) if nws.get("today_high") is not None: current_forecasts["NWS"] = _sf(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] = _sf(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 - + min(forecast_highs) if forecast_highs else forecast_high + forecast_median = ( + sorted(forecast_highs)[len(forecast_highs) // 2] if forecast_highs else None + ) + wind_speed = metar.get("current", {}).get("wind_speed_kt", 0) - + # 获取当地时间小时和分钟 local_time_full = open_meteo.get("current", {}).get("local_time", "") try: @@ -73,8 +78,9 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): 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: + 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 @@ -82,15 +88,28 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): # === DEB 融合渲染 === if city_name and current_forecasts: - blended_high, weight_info = calculate_dynamic_weights(city_name, current_forecasts) + blended_high, weight_info = calculate_dynamic_weights( + city_name, current_forecasts + ) if blended_high is not None: - insights.insert(0, f"🧬 DEB 融合预测：{blended_high}{temp_symbol} ({weight_info})") - ai_features.append(f"🧬 DEB系统已通过历史偏差矫正算出期待点是: {blended_high}{temp_symbol}。") - + insights.insert( + 0, + f"🧬 DEB 融合预测：{blended_high}{temp_symbol} ({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, deb_prediction=blended_high) - except: + update_daily_record( + city_name, + local_date_str, + current_forecasts, + max_so_far, + deb_prediction=blended_high, + ) + except Exception: pass # === METAR 趋势分析 (移到前部判断降温) === @@ -103,16 +122,31 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): diff = latest_val - prev_val if len(temps_only) >= 3: all_same = all(t == latest_val for t in temps_only[:3]) - all_rising = all(temps_only[i] >= temps_only[i+1] for i in range(min(3, len(temps_only)) - 1)) - all_falling = all(temps_only[i] <= temps_only[i+1] for i in range(min(3, len(temps_only)) - 1)) - trend_display = " → ".join([f"{t}{temp_symbol}@{tm}" for tm, t in recent_temps[:3]]) - if all_same: trend_desc = f"📉 温度已停滞（{trend_display}），大概率到顶。" - elif all_rising and diff > 0: trend_desc = f"📈 仍在升温（{trend_display}）。" - elif all_falling and diff < 0: trend_desc = f"📉 已开始降温（{trend_display}）。" - else: trend_desc = f"📊 温度波动中（{trend_display}）。" - elif diff == 0: trend_desc = f"📉 温度持平（最近两条都是 {latest_val}{temp_symbol}）。" - elif diff > 0: trend_desc = f"📈 仍在升温（{prev_val} → {latest_val}{temp_symbol}）。" - else: trend_desc = f"📉 已开始降温（{prev_val} → {latest_val}{temp_symbol}）。" + all_rising = all( + temps_only[i] >= temps_only[i + 1] + for i in range(min(3, len(temps_only)) - 1) + ) + all_falling = all( + temps_only[i] <= temps_only[i + 1] + for i in range(min(3, len(temps_only)) - 1) + ) + trend_display = " → ".join( + [f"{t}{temp_symbol}@{tm}" for tm, t in recent_temps[:3]] + ) + if all_same: + trend_desc = f"📉 温度已停滞（{trend_display}），大概率到顶。" + elif all_rising and diff > 0: + trend_desc = f"📈 仍在升温（{trend_display}）。" + elif all_falling and diff < 0: + trend_desc = f"📉 已开始降温（{trend_display}）。" + else: + trend_desc = f"📊 温度波动中（{trend_display}）。" + elif diff == 0: + trend_desc = f"📉 温度持平（最近两条都是 {latest_val}{temp_symbol}）。" + elif diff > 0: + trend_desc = f"📈 仍在升温（{prev_val} → {latest_val}{temp_symbol}）。" + else: + trend_desc = f"📉 已开始降温（{prev_val} → {latest_val}{temp_symbol}）。" is_cooling = "降温" in trend_desc @@ -140,11 +174,14 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): ens_median = _sf(ensemble.get("median")) if ens_p10 is not None and ens_p90 is not None and ens_median is not None: msg1 = f"📊 集合预报：中位数 {ens_median}{temp_symbol}，90% 区间 [{ens_p10}{temp_symbol} - {ens_p90}{temp_symbol}]。" - if not is_cooling: insights.append(msg1) + if not is_cooling: + insights.append(msg1) ai_features.append(msg1) if om_today is not None: - if om_today > ens_p90 and (max_so_far is None or max_so_far < om_today - 0.5): + if om_today > ens_p90 and ( + max_so_far is None or max_so_far < om_today - 0.5 + ): msg2 = f"⚡ 预报偏高：确定性预报 {om_today}{temp_symbol} 超集合90%上限，更可能接近 {ens_median}{temp_symbol}。" ai_features.append(msg2) elif om_today < ens_p10 and (max_so_far is None or max_so_far < ens_median): @@ -153,20 +190,23 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): # === 数学概率计算（基于集合预报正态分布拟合）=== import math as _math + # 用 P10/P90 反推标准差: P10 = median - 1.28*sigma, P90 = median + 1.28*sigma sigma = (ens_p90 - ens_p10) / 2.56 - if sigma < 0.1: sigma = 0.1 # 防止除以零 - + if sigma < 0.1: + sigma = 0.1 # 防止除以零 + # 用 DEB 历史 MAE 作为 σ 的下限 # 如果模型过去的平均误差远大于集合预报的 σ，说明集合低估了真实不确定性 if city_name: from src.analysis.deb_algorithm import get_deb_accuracy + acc = get_deb_accuracy(city_name) if acc: _, hist_mae, _, _ = acc if hist_mae > sigma: sigma = hist_mae - + # === Shock Score: 气象突变软评分 (0~1) === # 用近 4 条 METAR 的风向/云量/气压变化评估环境稳定性 # 越高 = 越不稳定 = σ 放宽 @@ -174,8 +214,8 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): recent_obs = metar.get("recent_obs", []) if len(recent_obs) >= 2: oldest = recent_obs[-1] # 最早 - newest = recent_obs[0] # 最新 - + newest = recent_obs[0] # 最新 + # ① 风向变化项 (0~0.4) # 角度差 × 风速放大系数（风速 > 10kt 时权重高，弱风忽略噪声） wdir_old = _sf(oldest.get("wdir")) @@ -183,11 +223,12 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): wspd_new = _sf(newest.get("wspd")) or 0 if wdir_old is not None and wdir_new is not None: angle_diff = abs(wdir_new - wdir_old) - if angle_diff > 180: angle_diff = 360 - angle_diff + if angle_diff > 180: + angle_diff = 360 - angle_diff wind_weight = min(wspd_new / 15.0, 1.0) wind_shock = min(angle_diff / 90.0, 1.0) * wind_weight * 0.4 shock_score += wind_shock - + # ② 云量阶跃项 (0~0.35) # CLR=0, FEW=1, SCT=2, BKN=3, OVC=4 cloud_old = oldest.get("cloud_rank", 0) @@ -195,7 +236,7 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): cloud_jump = abs(cloud_new - cloud_old) cloud_shock = min(cloud_jump / 3.0, 1.0) * 0.35 shock_score += cloud_shock - + # ③ 气压变化项 (0~0.25) # 2h 内气压变化超过 2hPa 视为异常 altim_old = _sf(oldest.get("altim")) @@ -204,11 +245,11 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): press_diff = abs(altim_new - altim_old) press_shock = min(press_diff / 4.0, 1.0) * 0.25 shock_score += press_shock - + # 应用 shock_score: 放宽 σ if shock_score > 0.05: - sigma *= (1 + 0.5 * shock_score) - + sigma *= 1 + 0.5 * shock_score + # 时间修正：根据当前时间距峰值的位置调整 σ # 峰值前：σ 不变（不确定性最大） # 峰值窗口内：σ 缩小 30%（正在定型） @@ -231,25 +272,34 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): if ens_p10 is not None and ens_p90 is not None and not is_dead_market: # (现有概率计算逻辑保留，但增加 is_dead_market 排除) - mu = forecast_median * 0.7 + ens_median * 0.3 if forecast_median is not None else ens_median + mu = ( + forecast_median * 0.7 + ens_median * 0.3 + if forecast_median is not None + else ens_median + ) if max_so_far is not None and max_so_far > mu: mu = max_so_far + (0.3 if not is_cooling else 0.0) - + def _norm_cdf(x, m, s): return 0.5 * (1 + _math.erf((x - m) / (s * _math.sqrt(2)))) - + min_possible_wu = round(max_so_far) if max_so_far is not None else -999 probs = {} for n in range(round(mu) - 2, round(mu) + 3): - if n < min_possible_wu: continue + if n < min_possible_wu: + continue p = _norm_cdf(n + 0.5, mu, sigma) - _norm_cdf(n - 0.5, mu, sigma) - if p > 0.01: probs[n] = p - + if p > 0.01: + probs[n] = p + total_p = sum(probs.values()) if total_p > 0: probs = {k: v / total_p for k, v in probs.items()} sorted_probs = sorted(probs.items(), key=lambda x: x[1], reverse=True) - prob_parts = [f"{int(t)}{temp_symbol} [{t-0.5}~{t+0.5}) {p*100:.0f}%" for t, p in sorted_probs[:4]] + prob_parts = [ + f"{int(t)}{temp_symbol} [{t - 0.5}~{t + 0.5}) {p * 100:.0f}%" + for t, p in sorted_probs[:4] + ] if prob_parts: prob_str = " | ".join(prob_parts) insights.append(f"🎲 结算概率 (μ={mu:.1f})：{prob_str}") @@ -258,7 +308,7 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): settled_wu = round(max_so_far) if max_so_far is not None else "N/A" dead_msg = f"🎲 结算预测：已锁定 {settled_wu}{temp_symbol} (死盘确认)" insights.append(dead_msg) - ai_features.append(f"🎲 状态: 确认死盘，结算已无悬念。") + ai_features.append("🎲 状态: 确认死盘，结算已无悬念。") # === 实测已超预报 & 趋势输出 === if max_so_far is not None and forecast_high is not None: @@ -266,8 +316,11 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): exceed_by = max_so_far - forecast_high bt_msg = f"🚨 实测已超预报：{max_so_far}{temp_symbol} 超过上限 {forecast_high}{temp_symbol}（+{exceed_by:.1f}°）。" insights.append(bt_msg) - ai_features.append(f"🚨 异常: 实测已冲破所有预报上限 ({max_so_far}{temp_symbol} vs {forecast_high}{temp_symbol})。") - if trend_desc: ai_features.append(trend_desc) + ai_features.append( + f"🚨 异常: 实测已冲破所有预报上限 ({max_so_far}{temp_symbol} vs {forecast_high}{temp_symbol})。" + ) + if trend_desc: + ai_features.append(trend_desc) else: if trend_desc: ai_features.append(trend_desc) @@ -289,21 +342,39 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): # === 峰值时刻 AI 提示 === if peak_hours: - window = f"{peak_hours[0]} - {peak_hours[-1]}" if len(peak_hours) > 1 else peak_hours[0] - + window = ( + f"{peak_hours[0]} - {peak_hours[-1]}" + if len(peak_hours) > 1 + else peak_hours[0] + ) + if local_hour <= last_peak_h: if last_peak_h < 6: - ai_features.append(f"⚠️ 提示：预测最热在凌晨，后续气温可能一路走低。") - elif local_hour < first_peak_h and (max_so_far is None or max_so_far < forecast_high): + ai_features.append( + "⚠️ 提示：预测最热在凌晨，后续气温可能一路走低。" + ) + elif local_hour < first_peak_h and ( + max_so_far is None or max_so_far < forecast_high + ): target_temp = om_today if om_today is not None else forecast_high - ai_features.append(f"🎯 关注重点：看看那个时段能否涨到 {target_temp}{temp_symbol}。") - + ai_features.append( + f"🎯 关注重点：看看那个时段能否涨到 {target_temp}{temp_symbol}。" + ) + # 写给AI（使用精确到分钟的时间） remain_hrs = first_peak_h - local_hour_frac - if local_hour_frac > last_peak_h: ai_features.append(f"⏱️ 状态: 预报峰值时段已过 ({window})。") - elif first_peak_h <= local_hour_frac <= last_peak_h: ai_features.append(f"⏱️ 状态: 正处于预报最热窗口 ({window})内。") - elif remain_hrs < 1: ai_features.append(f"⏱️ 状态: 距最热时段仅剩约 {int(remain_hrs * 60)} 分钟 ({window})。") - else: ai_features.append(f"⏱️ 状态: 距最热时段还有约 {remain_hrs:.1f}h ({window})。") + if local_hour_frac > last_peak_h: + ai_features.append(f"⏱️ 状态: 预报峰值时段已过 ({window})。") + elif first_peak_h <= local_hour_frac <= last_peak_h: + ai_features.append(f"⏱️ 状态: 正处于预报最热窗口 ({window})内。") + elif remain_hrs < 1: + ai_features.append( + f"⏱️ 状态: 距最热时段仅剩约 {int(remain_hrs * 60)} 分钟 ({window})。" + ) + else: + ai_features.append( + f"⏱️ 状态: 距最热时段还有约 {remain_hrs:.1f}h ({window})。" + ) # === 其他 AI 专供的事实特征 === # 明确告知 AI 当前实测温度和今日最高温，避免 AI 从趋势数据中误读 @@ -311,10 +382,13 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): if current_temp is not None: ai_features.append(f"🌡️ 当前实测温度: {current_temp}{temp_symbol}。") if max_so_far is not None: - ai_features.append(f"🏔️ 今日实测最高温: {max_so_far}{temp_symbol} (WU结算={round(max_so_far)}{temp_symbol})。") - + ai_features.append( + f"🏔️ 今日实测最高温: {max_so_far}{temp_symbol} (WU结算={round(max_so_far)}{temp_symbol})。" + ) + # 传递城市的 METAR 取整特性给 AI from src.data_collection.city_risk_profiles import get_city_risk_profile + if city_name: _profile = get_city_risk_profile(city_name) if _profile and _profile.get("metar_rounding"): @@ -323,16 +397,20 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): wind_dir = metar.get("current", {}).get("wind_dir", "未知") ai_features.append(f"🌬️ 当下风况: 约 {wind_speed}kt (方向 {wind_dir}°)。") humidity = metar.get("current", {}).get("humidity") - if humidity and humidity > 80: ai_features.append(f"💦 湿度极高 ({humidity}%)。") - + if humidity and humidity > 80: + ai_features.append(f"💦 湿度极高 ({humidity}%)。") + clouds = metar.get("current", {}).get("clouds", []) if clouds: cover = clouds[-1].get("cover", "") - c_desc = {"OVC": "全阴", "BKN": "多云", "SCT": "散云", "FEW": "少云"}.get(cover, cover) + c_desc = {"OVC": "全阴", "BKN": "多云", "SCT": "散云", "FEW": "少云"}.get( + cover, cover + ) ai_features.append(f"☁️ 天空状况: {c_desc}。") wx_desc = metar.get("current", {}).get("wx_desc") - if wx_desc: ai_features.append(f"🌧️ 天气现象: {wx_desc}。") + if wx_desc: + ai_features.append(f"🌧️ 天气现象: {wx_desc}。") max_temp_time_str = metar.get("current", {}).get("max_temp_time", "") if max_so_far is not None and max_temp_time_str: @@ -341,16 +419,23 @@ def analyze_weather_trend(weather_data, temp_symbol, city_name=None): max_temp_rad = 0.0 hourly_rad = hourly.get("shortwave_radiation", []) for t_str, rad in zip(times, hourly_rad): - if t_str.startswith(local_date_str) and int(t_str.split("T")[1][:2]) == max_h: + if ( + t_str.startswith(local_date_str) + and int(t_str.split("T")[1][:2]) == max_h + ): max_temp_rad = rad if rad is not None else 0.0 break if max_temp_rad < 50: - ai_features.append(f"🌙 动力事实: 最高温出现在低辐射时段 ({max_temp_time_str}, 辐射{max_temp_rad:.0f}W/m²)。") - except: pass + ai_features.append( + f"🌙 动力事实: 最高温出现在低辐射时段 ({max_temp_time_str}, 辐射{max_temp_rad:.0f}W/m²)。" + ) + except Exception: + pass display_str = "\n".join(insights) if insights else "" return display_str, "\n".join(ai_features) + def start_bot(): config = load_config() token = os.getenv("TELEGRAM_BOT_TOKEN") @@ -387,38 +472,51 @@ def start_bot(): try: parts = message.text.split(maxsplit=1) if len(parts) < 2: - bot.reply_to(message, "❓ 用法: /deb ankara", parse_mode="HTML") + bot.reply_to( + message, "❓ 用法: /deb ankara", parse_mode="HTML" + ) return - + city_input = parts[1].strip().lower() # 复用城市名映射 city_aliases = { - "ank": "ankara", "lon": "london", "par": "paris", - "nyc": "new york", "chi": "chicago", "dal": "dallas", - "mia": "miami", "atl": "atlanta", "sea": "seattle", - "tor": "toronto", "sel": "seoul", "ba": "buenos aires", + "ank": "ankara", + "lon": "london", + "par": "paris", + "nyc": "new york", + "chi": "chicago", + "dal": "dallas", + "mia": "miami", + "atl": "atlanta", + "sea": "seattle", + "tor": "toronto", + "sel": "seoul", + "ba": "buenos aires", "wel": "wellington", } city_name = city_aliases.get(city_input, city_input) - - from src.analysis.deb_algorithm import get_deb_accuracy, load_history + + from src.analysis.deb_algorithm import load_history import os as _os - + # 获取详细历史数据 project_root = _os.path.dirname(_os.path.abspath(__file__)) - history_file = _os.path.join(project_root, 'data', 'daily_records.json') + 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]: - bot.reply_to(message, f"❌ 暂无 {city_name} 的历史数据", parse_mode="HTML") + bot.reply_to( + message, f"❌ 暂无 {city_name} 的历史数据", parse_mode="HTML" + ) return - + city_data = data[city_name] from datetime import datetime as _dt + today_str = _dt.now().strftime("%Y-%m-%d") - + lines = [f"📊 DEB 准确率报告 - {city_name.title()}\n"] - + # 逐日明细 lines.append("📅 逐日记录：") total_days = 0 @@ -426,48 +524,59 @@ def start_bot(): deb_errors = [] signed_errors = [] # 有正负的误差 (DEB - 实测) model_errors = {} - + for date_str in sorted(city_data.keys()): record = city_data[date_str] - actual = record.get('actual_high') - deb_pred = record.get('deb_prediction') - forecasts = record.get('forecasts', {}) - + actual = record.get("actual_high") + deb_pred = record.get("deb_prediction") + forecasts = record.get("forecasts", {}) + if actual is None: continue try: actual = float(actual) - if deb_pred is not None: deb_pred = float(deb_pred) - except: continue - + if deb_pred is not None: + deb_pred = float(deb_pred) + except Exception: + continue + # 如果没有存 DEB 预测值，用当天各模型平均值回算 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 v in forecasts.values() if v is not None + ] if valid_preds: deb_pred = round(sum(valid_preds) / len(valid_preds), 1) - + actual_wu = round(actual) - + # DEB 命中判断 if deb_pred is not None and date_str != today_str: total_days += 1 deb_wu = round(deb_pred) hit = deb_wu == actual_wu - if hit: hits += 1 + if hit: + hits += 1 err = deb_pred - actual deb_errors.append(abs(err)) signed_errors.append(err) icon = "✅" if hit else "❌" - retro = "≈" if 'deb_prediction' not in record else "" + retro = "≈" if "deb_prediction" not in record else "" # 错误类型标签 if not hit: - err_label = f" 低估{abs(err):.1f}°" if err < 0 else f" 高估{abs(err):.1f}°" + err_label = ( + f" 低估{abs(err):.1f}°" + if err < 0 + else f" 高估{abs(err):.1f}°" + ) else: err_label = f" 偏差{abs(err):.1f}°" - lines.append(f" {date_str}: DEB {retro}{deb_pred}→{deb_wu} vs 实测 {actual}→{actual_wu} {icon}{err_label}") + lines.append( + f" {date_str}: DEB {retro}{deb_pred}→{deb_wu} vs 实测 {actual}→{actual_wu} {icon}{err_label}" + ) elif date_str == today_str: lines.append(f" {date_str}: 📍 今天进行中 (实测暂 {actual})") - + # 各模型误差统计 if date_str != today_str and actual is not None: for model, pred in forecasts.items(): @@ -475,53 +584,65 @@ def start_bot(): if model not in model_errors: model_errors[model] = [] model_errors[model].append(abs(float(pred) - actual)) - + # 汇总 if total_days > 0: hit_rate = hits / total_days * 100 deb_mae = sum(deb_errors) / len(deb_errors) - lines.append(f"\n🎯 DEB 总战绩：WU命中 {hits}/{total_days} ({hit_rate:.0f}%) | MAE: {deb_mae:.1f}°") - + lines.append( + f"\n🎯 DEB 总战绩：WU命中 {hits}/{total_days} ({hit_rate:.0f}%) | MAE: {deb_mae:.1f}°" + ) + # 和各模型 MAE 对比 if model_errors: - lines.append(f"\n📈 模型 MAE 对比：") - model_maes = {m: sum(e)/len(e) for m, e in model_errors.items() if e} + lines.append("\n📈 模型 MAE 对比：") + model_maes = { + m: sum(e) / len(e) for m, e in model_errors.items() if e + } sorted_models = sorted(model_maes.items(), key=lambda x: x[1]) for m, mae in sorted_models: tag = " ⭐" if mae <= deb_mae else "" lines.append(f" {m}: {mae:.1f}°{tag}") lines.append(f" DEB融合: {deb_mae:.1f}°") - + # 偏差模式分析 mean_bias = sum(signed_errors) / len(signed_errors) underest = sum(1 for e in signed_errors if e < -0.3) overest = sum(1 for e in signed_errors if e > 0.3) - - lines.append(f"\n🔍 偏差分析：") + + lines.append("\n🔍 偏差分析：") if abs(mean_bias) > 0.3: bias_dir = "低估" if mean_bias < 0 else "高估" lines.append(f" ⚠️ 系统性{bias_dir}：平均偏差 {mean_bias:+.1f}°") else: lines.append(f" ✅ 无明显系统偏差（平均 {mean_bias:+.1f}°）") - lines.append(f" 低估 {underest} 次 | 高估 {overest} 次 | 准确 {total_days - underest - overest} 次") - + lines.append( + f" 低估 {underest} 次 | 高估 {overest} 次 | 准确 {total_days - underest - overest} 次" + ) + # 可操作建议 - lines.append(f"\n💡 建议：") + lines.append("\n💡 建议：") if underest > overest and abs(mean_bias) > 0.5: - lines.append(f" 该城市模型集体低估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值高 {abs(mean_bias):.0f}-{abs(mean_bias)+0.5:.0f}°。交易时建议适当看高。") + lines.append( + f" 该城市模型集体低估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值高 {abs(mean_bias):.0f}-{abs(mean_bias) + 0.5:.0f}°。交易时建议适当看高。" + ) elif overest > underest and abs(mean_bias) > 0.5: - lines.append(f" 该城市模型集体高估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值低。交易时建议适当看低。") + lines.append( + f" 该城市模型集体高估趋势明显（{mean_bias:+.1f}°），实际最高温可能比 DEB 融合值低。交易时建议适当看低。" + ) elif deb_mae > 1.5: - lines.append(f" 该城市预报波动大 (MAE {deb_mae:.1f}°)，建议观望或轻仓。") + lines.append( + f" 该城市预报波动大 (MAE {deb_mae:.1f}°)，建议观望或轻仓。" + ) elif hit_rate >= 60: - lines.append(f" DEB 表现良好，可作为主要参考。") + lines.append(" DEB 表现良好，可作为主要参考。") else: - lines.append(f" 数据积累中，建议结合 AI 分析综合判断。") - - lines.append(f"\n📝 MAE = 平均绝对误差，越小越准。⭐ = 优于 DEB 融合。") + lines.append(" 数据积累中，建议结合 AI 分析综合判断。") + + lines.append("\n📝 MAE = 平均绝对误差，越小越准。⭐ = 优于 DEB 融合。") else: lines.append("\n⏳ 尚无完整的 DEB 预测记录，明天起开始统计。") - + bot.reply_to(message, "\n".join(lines), parse_mode="HTML") except Exception as e: bot.reply_to(message, f"❌ 查询失败: {e}") @@ -540,36 +661,52 @@ def start_bot(): return city_input = parts[1].strip().lower() - + # --- 核心标准名称映射表 --- # 这里的 Key 是缩写或别名，Value 是 Open-Meteo 识别的标准全称 STANDARD_MAPPING = { - "sel": "seoul", "seo": "seoul", "首尔": "seoul", - "lon": "london", "伦敦": "london", - "tor": "toronto", "多伦多": "toronto", - "ank": "ankara", "安卡拉": "ankara", - "wel": "wellington", "惠灵顿": "wellington", - "ba": "buenos aires", "布宜诺斯艾利斯": "buenos aires", - "nyc": "new york", "ny": "new york", "纽约": "new york", - "chi": "chicago", "芝加哥": "chicago", - "sea": "seattle", "西雅图": "seattle", - "mia": "miami", "迈阿密": "miami", - "atl": "atlanta", "亚特兰大": "atlanta", - "dal": "dallas", "达拉斯": "dallas", - "la": "los angeles", "洛杉矶": "los angeles", - "par": "paris", "巴黎": "paris", + "sel": "seoul", + "seo": "seoul", + "首尔": "seoul", + "lon": "london", + "伦敦": "london", + "tor": "toronto", + "多伦多": "toronto", + "ank": "ankara", + "安卡拉": "ankara", + "wel": "wellington", + "惠灵顿": "wellington", + "ba": "buenos aires", + "布宜诺斯艾利斯": "buenos aires", + "nyc": "new york", + "ny": "new york", + "纽约": "new york", + "chi": "chicago", + "芝加哥": "chicago", + "sea": "seattle", + "西雅图": "seattle", + "mia": "miami", + "迈阿密": "miami", + "atl": "atlanta", + "亚特兰大": "atlanta", + "dal": "dallas", + "达拉斯": "dallas", + "la": "los angeles", + "洛杉矶": "los angeles", + "par": "paris", + "巴黎": "paris", } - + # 支持的城市全名列表（用于模糊匹配） SUPPORTED_CITIES = list(set(STANDARD_MAPPING.values())) - + # 1. 第一优先级：严格全字匹配（别名/缩写） city_name = STANDARD_MAPPING.get(city_input) - + # 2. 第二优先级：输入本身就是城市全名 if not city_name and city_input in SUPPORTED_CITIES: city_name = city_input - + # 3. 第三优先级：前缀匹配（在别名和城市全名中搜索） if not city_name and len(city_input) >= 2: # 先搜别名 @@ -583,7 +720,7 @@ def start_bot(): if full_name.startswith(city_input): city_name = full_name break - + # 4. 未找到 → 报错，列出支持的城市 if not city_name: city_list = ", ".join(sorted(set(STANDARD_MAPPING.values()))) @@ -596,80 +733,105 @@ def start_bot(): ) return - bot.send_message(message.chat.id, f"🔍 正在查询 {city_name.title()} 的天气数据...") + bot.send_message( + message.chat.id, f"🔍 正在查询 {city_name.title()} 的天气数据..." + ) coords = weather.get_coordinates(city_name) if not coords: bot.reply_to(message, f"❌ 未找到城市坐标: {city_name}") return - weather_data = weather.fetch_all_sources(city_name, lat=coords["lat"], lon=coords["lon"]) + weather_data = weather.fetch_all_sources( + city_name, lat=coords["lat"], lon=coords["lon"] + ) open_meteo = weather_data.get("open-meteo", {}) metar = weather_data.get("metar", {}) mgm = weather_data.get("mgm", {}) - + # 数值归一化 def _sf(v): - if v is None: return None - try: return float(v) - except: return None - + if v is None: + return None + try: + return float(v) + except Exception: + return None + temp_unit = open_meteo.get("unit", "celsius") temp_symbol = "°F" if temp_unit == "fahrenheit" else "°C" - + # --- 1. 紧凑 Header (城市 + 时间 + 风险状态) --- local_time = open_meteo.get("current", {}).get("local_time", "") time_str = local_time.split(" ")[1][:5] if " " in local_time else "N/A" - + risk_profile = get_city_risk_profile(city_name) risk_emoji = risk_profile.get("risk_level", "⚪") if risk_profile else "⚪" - + msg_header = f"📍 {city_name.title()} ({time_str}) {risk_emoji}" msg_lines = [msg_header] - + # --- 2. 紧凑 风险提示 --- if risk_profile: bias = risk_profile.get("bias", "±0.0") - msg_lines.append(f"⚠️ {risk_profile.get('airport_name', '')}: {bias}{temp_symbol} | {risk_profile.get('warning', '')}") + msg_lines.append( + f"⚠️ {risk_profile.get('airport_name', '')}: {bias}{temp_symbol} | {risk_profile.get('warning', '')}" + ) # --- 3. 紧凑 预测区 --- daily = open_meteo.get("daily", {}) dates = daily.get("time", [])[:3] max_temps = daily.get("temperature_2m_max", [])[:3] - + nws_high = _sf(weather_data.get("nws", {}).get("today_high")) mgm_high = _sf(mgm.get("today_high")) mb_high = _sf(weather_data.get("meteoblue", {}).get("today_high")) - + # 今天对比 today_t = max_temps[0] if max_temps else "N/A" comp_parts = [] sources = ["Open-Meteo"] - + if mb_high is not None: sources.append("MB") - comp_parts.append(f"MB: {mb_high:.1f}{temp_symbol}" if isinstance(mb_high, (int, float)) else f"MB: {mb_high}") + comp_parts.append( + f"MB: {mb_high:.1f}{temp_symbol}" + if isinstance(mb_high, (int, float)) + else f"MB: {mb_high}" + ) if nws_high is not None: sources.append("NWS") - comp_parts.append(f"NWS: {nws_high:.1f}{temp_symbol}" if isinstance(nws_high, (int, float)) else f"NWS: {nws_high}") + comp_parts.append( + f"NWS: {nws_high:.1f}{temp_symbol}" + if isinstance(nws_high, (int, float)) + else f"NWS: {nws_high}" + ) if mgm_high is not None: sources.append("MGM") - comp_parts.append(f"MGM: {mgm_high:.1f}{temp_symbol}" if isinstance(mgm_high, (int, float)) else f"MGM: {mgm_high}") - + comp_parts.append( + f"MGM: {mgm_high:.1f}{temp_symbol}" + if isinstance(mgm_high, (int, float)) + else f"MGM: {mgm_high}" + ) + # 检查是否有显著分歧 (超过 5°F 或 2.5°C) divergence_warning = "" if mb_high is not None and max_temps: diff = abs(mb_high - (_sf(max_temps[0]) or 0)) threshold = 5.0 if temp_unit == "fahrenheit" else 2.5 if diff > threshold: - divergence_warning = f" ⚠️ 模型显著分歧 ({diff:.1f}{temp_symbol})" - + divergence_warning = ( + f" ⚠️ 模型显著分歧 ({diff:.1f}{temp_symbol})" + ) + comp_str = f" ({' | '.join(comp_parts)})" if comp_parts else "" sources_str = " | ".join(sources) - + msg_lines.append(f"\n📊 预报 ({sources_str})") - msg_lines.append(f"👉 今天: {today_t}{temp_symbol}{comp_str}{divergence_warning}") - + msg_lines.append( + f"👉 今天: {today_t}{temp_symbol}{comp_str}{divergence_warning}" + ) + # 明后天 if len(dates) > 1: future_forecasts = [] @@ -682,8 +844,16 @@ def start_bot(): sunsets = daily.get("sunset", []) sunshine_durations = daily.get("sunshine_duration", []) if sunrises and sunsets: - sunrise_t = sunrises[0].split("T")[1][:5] if "T" in str(sunrises[0]) else sunrises[0] - sunset_t = sunsets[0].split("T")[1][:5] if "T" in str(sunsets[0]) else sunsets[0] + sunrise_t = ( + sunrises[0].split("T")[1][:5] + if "T" in str(sunrises[0]) + else sunrises[0] + ) + sunset_t = ( + sunsets[0].split("T")[1][:5] + if "T" in str(sunsets[0]) + else sunsets[0] + ) sun_line = f"🌅 日出 {sunrise_t} | 🌇 日落 {sunset_t}" if sunshine_durations: sunshine_hours = sunshine_durations[0] / 3600 # 秒 -> 小时 @@ -692,21 +862,32 @@ def start_bot(): # --- 4. 核心 实测区 (合并 METAR 和 MGM) --- # 基础数据优先用 METAR - cur_temp = _sf(metar.get("current", {}).get("temp") if metar else mgm.get("current", {}).get("temp")) - max_p = _sf(metar.get("current", {}).get("max_temp_so_far") if metar else None) - max_p_time = metar.get("current", {}).get("max_temp_time") if metar else None + cur_temp = _sf( + metar.get("current", {}).get("temp") + if metar + else mgm.get("current", {}).get("temp") + ) + max_p = _sf( + metar.get("current", {}).get("max_temp_so_far") if metar else None + ) + max_p_time = ( + metar.get("current", {}).get("max_temp_time") if metar else None + ) obs_t_str = "N/A" metar_age_min = None # METAR 数据年龄（分钟） main_source = "METAR" if metar else "MGM" - + if metar: obs_t = metar.get("observation_time", "") try: if "T" in obs_t: from datetime import datetime, timezone, timedelta + dt = datetime.fromisoformat(obs_t.replace("Z", "+00:00")) utc_offset = open_meteo.get("utc_offset", 0) - local_dt = dt.astimezone(timezone(timedelta(seconds=utc_offset))) + local_dt = dt.astimezone( + timezone(timedelta(seconds=utc_offset)) + ) obs_t_str = local_dt.strftime("%H:%M") # 计算数据年龄 now_utc = datetime.now(timezone.utc) @@ -715,14 +896,17 @@ def start_bot(): obs_t_str = obs_t.split(" ")[1][:5] else: obs_t_str = obs_t - except: + except Exception: obs_t_str = obs_t[:16] elif mgm: m_time = mgm.get("current", {}).get("time", "") if "T" in m_time: from datetime import datetime, timezone, timedelta + dt = datetime.fromisoformat(m_time.replace("Z", "+00:00")) - m_time = dt.astimezone(timezone(timedelta(hours=3))).strftime("%H:%M") + m_time = dt.astimezone(timezone(timedelta(hours=3))).strftime( + "%H:%M" + ) elif " " in m_time: m_time = m_time.split(" ")[1][:5] obs_t_str = m_time @@ -738,6 +922,7 @@ def start_bot(): max_str = "" if max_p is not None: import math + settled_val = math.floor(max_p + 0.5) max_str = f" (最高: {max_p}{temp_symbol}" if max_p_time: @@ -754,7 +939,17 @@ def start_bot(): if metar_wx: wx_upper = metar_wx.upper().strip() wx_tokens = set(wx_upper.split()) - rain_codes = {"RA", "DZ", "-RA", "+RA", "-DZ", "+DZ", "TSRA", "SHRA", "FZRA"} + rain_codes = { + "RA", + "DZ", + "-RA", + "+RA", + "-DZ", + "+DZ", + "TSRA", + "SHRA", + "FZRA", + } snow_codes = {"SN", "GR", "GS", "-SN", "+SN", "BLSN"} fog_codes = {"FG", "BR", "HZ", "FZFG"} ts_codes = {"TS", "TSRA"} @@ -763,7 +958,9 @@ def start_bot(): elif {"+RA", "+SN"} & wx_tokens: wx_summary = "🌧️ 大雨" if "+RA" in wx_tokens else "❄️ 大雪" elif rain_codes & wx_tokens: - wx_summary = "🌧️ 小雨" if {"-RA", "-DZ", "DZ"} & wx_tokens else "🌧️ 下雨" + wx_summary = ( + "🌧️ 小雨" if {"-RA", "-DZ", "DZ"} & wx_tokens else "🌧️ 下雨" + ) elif snow_codes & wx_tokens: wx_summary = "❄️ 下雪" elif fog_codes & wx_tokens: @@ -775,26 +972,45 @@ def start_bot(): cover_code = "" if metar_clouds: cover_code = metar_clouds[-1].get("cover", "") - - if cover_code in ("SKC", "CLR") or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 1): + + if cover_code in ("SKC", "CLR") or ( + cover_code == "" and mgm_cloud is not None and mgm_cloud <= 1 + ): wx_summary = "☀️ 晴" - elif cover_code == "FEW" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 2): + elif cover_code == "FEW" or ( + cover_code == "" and mgm_cloud is not None and mgm_cloud <= 2 + ): wx_summary = "🌤️ 晴间少云" - elif cover_code == "SCT" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 4): + elif cover_code == "SCT" or ( + cover_code == "" and mgm_cloud is not None and mgm_cloud <= 4 + ): wx_summary = "⛅ 晴间多云" - elif cover_code == "BKN" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 6): + elif cover_code == "BKN" or ( + cover_code == "" and mgm_cloud is not None and mgm_cloud <= 6 + ): wx_summary = "🌥️ 多云" - elif cover_code == "OVC" or (cover_code == "" and mgm_cloud is not None and mgm_cloud <= 8): + elif cover_code == "OVC" or ( + cover_code == "" and mgm_cloud is not None and mgm_cloud <= 8 + ): wx_summary = "☁️ 阴天" elif mgm_cloud is not None: - cloud_names = {0: "☀️ 晴", 1: "🌤️ 晴", 2: "🌤️ 少云", 3: "⛅ 散云", 4: "⛅ 散云", 5: "🌥️ 多云", 6: "🌥️ 多云", 7: "☁️ 阴", 8: "☁️ 阴天"} + cloud_names = { + 0: "☀️ 晴", + 1: "🌤️ 晴", + 2: "🌤️ 少云", + 3: "⛅ 散云", + 4: "⛅ 散云", + 5: "🌥️ 多云", + 6: "🌥️ 多云", + 7: "☁️ 阴", + 8: "☁️ 阴天", + } wx_summary = cloud_names.get(mgm_cloud, "") wx_display = f" {wx_summary}" if wx_summary else "" - msg_lines.append(f"\n✈️ 实测 ({main_source}): {cur_temp}{temp_symbol}{max_str} |{wx_display} | {obs_t_str}{age_tag}") - - - + msg_lines.append( + f"\n✈️ 实测 ({main_source}): {cur_temp}{temp_symbol}{max_str} |{wx_display} | {obs_t_str}{age_tag}" + ) if mgm: m_c = mgm.get("current", {}) @@ -805,20 +1021,24 @@ def start_bot(): if wind_dir is not None: dirs = ["北", "东北", "东", "东南", "南", "西南", "西", "西北"] dir_str = dirs[int((float(wind_dir) + 22.5) % 360 / 45)] + "风 " - + # 体感和湿度（跳过缺失数据） feels_like = m_c.get("feels_like") humidity = m_c.get("humidity") if feels_like is not None or humidity is not None: parts = [] - if feels_like is not None: parts.append(f"🌡️ 体感: {feels_like}°C") - if humidity is not None: parts.append(f"💧 {humidity}%") + if feels_like is not None: + parts.append(f"🌡️ 体感: {feels_like}°C") + if humidity is not None: + parts.append(f"💧 {humidity}%") msg_lines.append(f" [MGM] {' | '.join(parts)}") - + # 风况（跳过缺失数据） if wind_dir is not None and wind_speed_ms is not None: - msg_lines.append(f" [MGM] 🌬️ {dir_str}{wind_dir}° ({wind_speed_ms} m/s) | 💧 降水: {m_c.get('rain_24h') or 0}mm") - + msg_lines.append( + f" [MGM] 🌬️ {dir_str}{wind_dir}° ({wind_speed_ms} m/s) | 💧 降水: {m_c.get('rain_24h') or 0}mm" + ) + # 新增：气压和云量 extra_parts = [] pressure = m_c.get("pressure") @@ -826,7 +1046,17 @@ def start_bot(): extra_parts.append(f"🌡 气压: {pressure}hPa") cloud_cover = m_c.get("cloud_cover") if cloud_cover is not None: - cloud_desc_map = {0: "晴朗", 1: "少云", 2: "少云", 3: "散云", 4: "散云", 5: "多云", 6: "多云", 7: "很多云", 8: "阴天"} + cloud_desc_map = { + 0: "晴朗", + 1: "少云", + 2: "少云", + 3: "散云", + 4: "散云", + 5: "多云", + 6: "多云", + 7: "很多云", + 8: "阴天", + } cloud_text = cloud_desc_map.get(cloud_cover, f"{cloud_cover}/8") extra_parts.append(f"☁️ 云量: {cloud_text}({cloud_cover}/8)") mgm_max = m_c.get("mgm_max_temp") @@ -834,33 +1064,46 @@ def start_bot(): extra_parts.append(f"🌡️ MGM最高: {mgm_max}°C") if extra_parts: msg_lines.append(f" [MGM] {' | '.join(extra_parts)}") - + if metar: m_c = metar.get("current", {}) wind = m_c.get("wind_speed_kt") wind_dir = m_c.get("wind_dir") vis = m_c.get("visibility_mi") clouds = m_c.get("clouds", []) - + cloud_desc = "" if clouds: - c_map = {"BKN": "多云", "OVC": "阴天", "FEW": "少云", "SCT": "散云", "SKC": "晴", "CLR": "晴"} + c_map = { + "BKN": "多云", + "OVC": "阴天", + "FEW": "少云", + "SCT": "散云", + "SKC": "晴", + "CLR": "晴", + } main = clouds[-1] cloud_desc = f"☁️ {c_map.get(main.get('cover'), main.get('cover'))}" prefix = "[METAR]" if mgm else " " if not mgm: - msg_lines.append(f" {prefix} 💨 {wind or 0}kt ({wind_dir or 0}°) | 👁️ {vis or 10}mi") - + msg_lines.append( + f" {prefix} 💨 {wind or 0}kt ({wind_dir or 0}°) | 👁️ {vis or 10}mi" + ) + if cloud_desc: - msg_lines.append(f" {prefix} {cloud_desc} | 👁️ {vis or 10}mi | 💨 {wind or 0}kt") + 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, city_name) + feature_str, ai_context = analyze_weather_trend( + weather_data, temp_symbol, city_name + ) if feature_str: # 仅将最核心的信息展示给用户作为"态势分析" # 但后面会把更全的数据传给 AI - msg_lines.append(f"\n💡 分析:") + msg_lines.append("\n💡 分析:") for line in feature_str.split("\n"): if line.strip(): msg_lines.append(f"- {line.strip()}") @@ -869,11 +1112,17 @@ def start_bot(): try: from src.analysis.ai_analyzer import get_ai_analysis # 构建更全的背景数据给 AI - + # 补充多模型分歧 mm = weather_data.get("multi_model", {}) if mm.get("forecasts"): - mm_str = " | ".join([f"{k}:{v}{temp_symbol}" for k,v in mm["forecasts"].items() if v]) + mm_str = " | ".join( + [ + f"{k}:{v}{temp_symbol}" + for k, v in mm["forecasts"].items() + if v + ] + ) ai_context += f"\n模型分歧: {mm_str}" ai_result = get_ai_analysis(ai_context, city_name, temp_symbol) @@ -886,11 +1135,13 @@ def start_bot(): except Exception as e: import traceback + logger.error(f"查询失败: {e}\n{traceback.format_exc()}") bot.reply_to(message, f"❌ 查询失败: {e}") logger.info("🤖 Bot 启动中...") bot.infinity_polling() + if __name__ == "__main__": start_bot() diff --git a/src/analysis/ai_analyzer.py b/src/analysis/ai_analyzer.py index e9f29c46..ca04e0b5 100644 --- a/src/analysis/ai_analyzer.py +++ b/src/analysis/ai_analyzer.py @@ -9,6 +9,7 @@ MODELS = [ "llama-3.1-8b-instant", ] + def get_ai_analysis(weather_insights: str, city_name: str, temp_symbol: str) -> str: """ 通过 Groq API (LLaMA 3.3 70B) 对天气态势进行极速交易分析 @@ -18,13 +19,10 @@ def get_ai_analysis(weather_insights: str, city_name: str, temp_symbol: str) -> if not api_key: logger.warning("GROQ_API_KEY 未配置，跳过 AI 分析") return "" - + url = "https://api.groq.com/openai/v1/chat/completions" - headers = { - "Authorization": f"Bearer {api_key}", - "Content-Type": "application/json" - } - + headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} + prompt = f""" 你是一个专业的天气衍生品（如 Polymarket）交易员。你的任务是分析当前天气特征，判断今日实测最高温是否能达到或超过预报中的【最高值】。 @@ -69,23 +67,26 @@ P4 **预报背景**（最低优先级）： payload = { "model": model, "messages": [ - {"role": "system", "content": "你是不讲废话、只看数据的专业气象分析师。"}, - {"role": "user", "content": prompt} + { + "role": "system", + "content": "你是不讲废话、只看数据的专业气象分析师。", + }, + {"role": "user", "content": prompt}, ], "temperature": 0.5, - "max_tokens": 250 + "max_tokens": 250, } response = requests.post(url, json=payload, headers=headers, timeout=15) response.raise_for_status() - + result = response.json() - content = result['choices'][0]['message']['content'].strip() - + content = result["choices"][0]["message"]["content"].strip() + if model != MODELS[0]: logger.info(f"Groq 降级到备用模型 {model} 成功") return content - + except requests.exceptions.HTTPError as e: status = e.response.status_code if e.response is not None else 0 if status in (500, 502, 503) and attempt == 0: @@ -93,7 +94,9 @@ P4 **预报背景**（最低优先级）： time.sleep(1.5) continue else: - logger.warning(f"Groq {model} 失败 (HTTP {status})，尝试下一个模型...") + logger.warning( + f"Groq {model} 失败 (HTTP {status})，尝试下一个模型..." + ) break # 换下一个模型 except Exception as e: logger.warning(f"Groq {model} 异常: {e}，尝试下一个模型...") @@ -101,4 +104,3 @@ P4 **预报背景**（最低优先级）： logger.error("所有 Groq 模型均不可用") return "\n⚠️ Groq AI 暂时不可用，请稍后再试" - diff --git a/src/analysis/deb_algorithm.py b/src/analysis/deb_algorithm.py index 9f251516..747654d2 100644 --- a/src/analysis/deb_algorithm.py +++ b/src/analysis/deb_algorithm.py @@ -1,6 +1,5 @@ import os import json -import logging from datetime import datetime, timedelta import fcntl @@ -9,23 +8,24 @@ import fcntl _history_cache = {} _history_mtime = 0 + def load_history(filepath): global _history_cache, _history_mtime if not os.path.exists(filepath): return {} - + try: current_mtime = os.path.getmtime(filepath) if current_mtime == _history_mtime and _history_cache: return _history_cache - - with open(filepath, 'r', encoding='utf-8') as f: + + with open(filepath, "r", encoding="utf-8") as f: # We don't strictly need a lock for reading in Python if the write is atomic, # but using one prevents reading half-written JSONs. fcntl.flock(f, fcntl.LOCK_SH) data = json.load(f) fcntl.flock(f, fcntl.LOCK_UN) - + _history_cache = data _history_mtime = current_mtime return data @@ -33,11 +33,12 @@ def load_history(filepath): print(f"Error loading history: {e}") return _history_cache if _history_cache else {} + def save_history(filepath, data): global _history_cache, _history_mtime _history_cache = data try: - with open(filepath, 'w', encoding='utf-8') as f: + with open(filepath, "w", encoding="utf-8") as f: fcntl.flock(f, fcntl.LOCK_EX) json.dump(data, f, ensure_ascii=False, indent=2) fcntl.flock(f, fcntl.LOCK_UN) @@ -45,94 +46,106 @@ def save_history(filepath, data): except Exception as e: print(f"Error saving history: {e}") -def update_daily_record(city_name, date_str, forecasts, actual_high, deb_prediction=None): + +def update_daily_record( + city_name, date_str, forecasts, actual_high, deb_prediction=None +): """ 保存/更新某城市某天的各个模型预报与最终实测值 forecasts: dict, 例如 {"ECMWF": 28.5, "GFS": 30.0, ...} actual_high: float, 最终实测最高温 deb_prediction: float, DEB 融合预测值（用于准确率追踪） """ - 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') - + 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] = {} - + # 避免无意义的频繁磁盘写入 - old_actual = data[city_name][date_str].get('actual_high') - if old_actual == actual_high and data[city_name][date_str].get('forecasts') == forecasts: + old_actual = data[city_name][date_str].get("actual_high") + if ( + old_actual == actual_high + and data[city_name][date_str].get("forecasts") == forecasts + ): return - - data[city_name][date_str]['forecasts'] = forecasts - data[city_name][date_str]['actual_high'] = actual_high + + data[city_name][date_str]["forecasts"] = forecasts + data[city_name][date_str]["actual_high"] = actual_high if deb_prediction is not None: - data[city_name][date_str]['deb_prediction'] = deb_prediction - + data[city_name][date_str]["deb_prediction"] = deb_prediction + # 自动清理：只保留最近 14 天的记录（DEB 只用 7 天，14 天留足余量） cutoff = (datetime.now() - timedelta(days=14)).strftime("%Y-%m-%d") for city in list(data.keys()): old_dates = [d for d in data[city] if d < cutoff] for d in old_dates: del data[city][d] - + 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') + 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, "暂无模型数据" + 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', {}) - + 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(): @@ -140,28 +153,32 @@ def calculate_dynamic_weights(city_name, current_forecasts, lookback_days=7): maes[model] = sum(err_list) / len(err_list) else: # 如果某个新模型没有历史数据，给它一个平均误差 - maes[model] = 2.0 - + 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} - + 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}°)") - + weight_str_parts.append(f"{m}({w * 100:.0f}%,MAE:{maes[m]:.1f}°)") + return round(blended_high, 1), " | ".join(weight_str_parts) @@ -174,49 +191,53 @@ def get_deb_accuracy(city_name): - total_days: 有效天数 - details_str: 格式化的展示字符串 """ - 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') + 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: return None - + city_data = data[city_name] today_str = datetime.now().strftime("%Y-%m-%d") - + hits = 0 total = 0 errors = [] - + for date_str in sorted(city_data.keys()): if date_str == today_str: continue # 跳过今天，还没结算 record = city_data[date_str] - deb_pred = record.get('deb_prediction') - actual = record.get('actual_high') - + deb_pred = record.get("deb_prediction") + actual = record.get("actual_high") + if deb_pred is None or actual is None: continue - + try: deb_pred = float(deb_pred) actual = float(actual) - except: + except Exception: continue - + total += 1 deb_wu = round(deb_pred) actual_wu = round(actual) if deb_wu == actual_wu: hits += 1 errors.append(abs(deb_pred - actual)) - + if total == 0: return None - + hit_rate = hits / total * 100 mae = sum(errors) / len(errors) - - details_str = f"过去{total}天 WU命中 {hits}/{total} ({hit_rate:.0f}%) | MAE: {mae:.1f}°" - + + details_str = ( + f"过去{total}天 WU命中 {hits}/{total} ({hit_rate:.0f}%) | MAE: {mae:.1f}°" + ) + return hit_rate, mae, total, details_str diff --git a/src/data_collection/city_risk_profiles.py b/src/data_collection/city_risk_profiles.py index 3e8ce1a6..3ee6bf7a 100644 --- a/src/data_collection/city_risk_profiles.py +++ b/src/data_collection/city_risk_profiles.py @@ -27,7 +27,6 @@ CITY_RISK_PROFILES = { "warning": "冬天温差最不稳定", "season_notes": "冬季", }, - # 🟡 中危城市 - 存在系统偏差，需注意 "ankara": { "risk_level": "medium", @@ -90,7 +89,6 @@ CITY_RISK_PROFILES = { "warning": "机场在北郊，冬季北风时比市区更冷", "season_notes": "夏季热浪期间偏差最大", }, - # 🟢 低危城市 - 数据相对靠谱 "toronto": { "risk_level": "low", @@ -170,7 +168,7 @@ CITY_RISK_PROFILES = { def get_city_risk_profile(city_name: str) -> dict: """获取城市的风险档案""" city_lower = city_name.lower().strip() - + city_key = city_lower return CITY_RISK_PROFILES.get(city_key) @@ -179,32 +177,28 @@ def format_risk_warning(profile: dict, temp_symbol: str) -> str: """格式化风险警告信息""" if not profile: return "" - + lines = [] - + # 风险等级标题 - risk_labels = { - "high": "高危", - "medium": "中危", - "low": "低危" - } + risk_labels = {"high": "高危", "medium": "中危", "low": "低危"} risk_label = risk_labels.get(profile["risk_level"], "未知") lines.append(f"⚠️ 数据偏差风险: {profile['risk_emoji']} {risk_label}") - + # 机场信息 lines.append(f" 📍 机场: {profile['airport_name']} ({profile['icao']})") lines.append(f" 📏 距市区: {profile['distance_km']}km") - + # 典型偏差 if profile["typical_bias_f"] >= 1.0: lines.append(f" 📊 偏差: ±{profile['typical_bias_f']}{temp_symbol}") - + # 偏差方向说明 if profile["bias_direction"]: lines.append(f" 💡 {profile['bias_direction']}") - + # 特别警告 if profile["warning"]: lines.append(f" 🚨 {profile['warning']}") - + return "\n".join(lines) diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 27ed3920..2fa70798 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -207,7 +207,9 @@ class WeatherDataCollector: return None - def fetch_metar(self, city: str, use_fahrenheit: bool = False, utc_offset: int = 0) -> Optional[Dict]: + def fetch_metar( + self, city: str, use_fahrenheit: bool = False, utc_offset: int = 0 + ) -> Optional[Dict]: """ 从 NOAA Aviation Weather Center 获取 METAR 航空气象数据 @@ -236,10 +238,10 @@ class WeatherDataCollector: } response = self.session.get( - url, - params=params, + url, + params=params, headers={"Cache-Control": "no-cache", "Pragma": "no-cache"}, - timeout=self.timeout + timeout=self.timeout, ) response.raise_for_status() @@ -251,46 +253,60 @@ class WeatherDataCollector: latest = data[0] temp_c = latest.get("temp") dewp_c = latest.get("dewp") - + # 从 rawOb 中提取真实观测时间（比 reportTime 更准确，reportTime 会被取整） # rawOb 格式: "METAR EGLC 271150Z AUTO ..." → "271150Z" → 27日11:50 UTC def _parse_rawob_time(obs): """从 rawOb 中提取精确的 UTC 观测时间""" raw = obs.get("rawOb", "") import re as _re - m = _re.search(r'\b(\d{2})(\d{2})(\d{2})Z\b', raw) + + m = _re.search(r"\b(\d{2})(\d{2})(\d{2})Z\b", raw) if m: - day, hour, minute = int(m.group(1)), int(m.group(2)), int(m.group(3)) + _day, hour, minute = ( + int(m.group(1)), + int(m.group(2)), + int(m.group(3)), + ) # 用 reportTime 的日期部分 + rawOb 的时分 fallback = obs.get("reportTime", "") try: clean = fallback.replace(" ", "T") - if not clean.endswith("Z"): clean += "Z" + if not clean.endswith("Z"): + clean += "Z" base_dt = datetime.fromisoformat(clean.replace("Z", "+00:00")) result = base_dt.replace(hour=hour, minute=minute, second=0) # 处理跨日（如 rawOb 是23:50但 reportTime 已经是次日00:00） if result > base_dt + timedelta(hours=2): result -= timedelta(days=1) return result - except: + except Exception: pass # fallback 到 reportTime fallback = obs.get("reportTime", "") try: clean = fallback.replace(" ", "T") - if not clean.endswith("Z"): clean += "Z" + if not clean.endswith("Z"): + clean += "Z" return datetime.fromisoformat(clean.replace("Z", "+00:00")) - except: + except Exception: return None - + obs_dt = _parse_rawob_time(latest) - obs_time = obs_dt.strftime("%Y-%m-%dT%H:%M:%S.000Z") if obs_dt else latest.get("reportTime", "") + obs_time = ( + obs_dt.strftime("%Y-%m-%dT%H:%M:%S.000Z") + if obs_dt + else latest.get("reportTime", "") + ) # 2. 精确计算"当地今天"的最高温 from datetime import timezone, timedelta + now_utc = datetime.now(timezone.utc) local_now = now_utc + timedelta(seconds=utc_offset) - local_midnight = local_now.replace(hour=0, minute=0, second=0, microsecond=0) + local_midnight = local_now.replace( + hour=0, minute=0, second=0, microsecond=0 + ) utc_midnight = local_midnight - timedelta(seconds=utc_offset) max_so_far_c = -999 @@ -306,12 +322,12 @@ class WeatherDataCollector: max_so_far_c = t local_report = obs_dt_iter + timedelta(seconds=utc_offset) max_temp_time = local_report.strftime("%H:%M") - except: + except Exception: continue # 3. 提取最近 4 条报文的多维数据（温度 + 风/云/压强，用于趋势和 shock_score） recent_temps_raw = [] # [(local_time_str, temp_c), ...] - recent_obs_raw = [] # [{time, temp, wdir, wspd, clouds, altim}, ...] + recent_obs_raw = [] # [{time, temp, wdir, wspd, clouds, altim}, ...] for obs in data[:4]: # data 已按时间倒序 obs_temp = obs.get("temp") obs_dt_iter = _parse_rawob_time(obs) @@ -319,21 +335,30 @@ class WeatherDataCollector: local_rt = obs_dt_iter + timedelta(seconds=utc_offset) recent_temps_raw.append((local_rt.strftime("%H:%M"), obs_temp)) # 云量码映射: CLR=0, FEW=1, SCT=2, BKN=3, OVC=4 - cloud_rank_map = {"CLR": 0, "SKC": 0, "FEW": 1, "SCT": 2, "BKN": 3, "OVC": 4} + cloud_rank_map = { + "CLR": 0, + "SKC": 0, + "FEW": 1, + "SCT": 2, + "BKN": 3, + "OVC": 4, + } clouds = obs.get("clouds", []) max_cloud_rank = 0 for c in clouds: rank = cloud_rank_map.get(c.get("cover", ""), 0) if rank > max_cloud_rank: max_cloud_rank = rank - recent_obs_raw.append({ - "time": local_rt.strftime("%H:%M"), - "temp": obs_temp, - "wdir": obs.get("wdir"), - "wspd": obs.get("wspd"), - "cloud_rank": max_cloud_rank, # 0~4 - "altim": obs.get("altim"), - }) + recent_obs_raw.append( + { + "time": local_rt.strftime("%H:%M"), + "temp": obs_temp, + "wdir": obs.get("wdir"), + "wspd": obs.get("wspd"), + "cloud_rank": max_cloud_rank, # 0~4 + "altim": obs.get("altim"), + } + ) # 转换为单位 if use_fahrenheit: @@ -342,7 +367,9 @@ class WeatherDataCollector: dewp = dewp_c * 9 / 5 + 32 if dewp_c is not None else None unit = "fahrenheit" # 转换最近温度 - recent_temps = [(t, round(v * 9 / 5 + 32, 1)) for t, v in recent_temps_raw] + recent_temps = [ + (t, round(v * 9 / 5 + 32, 1)) for t, v in recent_temps_raw + ] else: temp = temp_c max_so_far = max_so_far_c if max_so_far_c > -900 else None @@ -358,7 +385,9 @@ class WeatherDataCollector: "observation_time": obs_time, "current": { "temp": round(temp, 1) if temp is not None else None, - "max_temp_so_far": round(max_so_far, 1) if max_so_far is not None else None, + "max_temp_so_far": round(max_so_far, 1) + if max_so_far is not None + else None, "max_temp_time": max_temp_time, "dewpoint": round(dewp, 1) if dewp is not None else None, "humidity": latest.get("rh"), @@ -396,17 +425,18 @@ class WeatherDataCollector: # 必须带 Origin，否则会被反爬拦截 headers = { "Origin": "https://www.mgm.gov.tr", - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", } results = {} - + try: # 1. 实时数据 (添加时间戳防止 CDN 缓存) import time + obs_resp = self.session.get( - f"{base_url}/sondurumlar?istno={istno}&_={int(time.time()*1000)}", - headers=headers, - timeout=self.timeout + f"{base_url}/sondurumlar?istno={istno}&_={int(time.time() * 1000)}", + headers=headers, + timeout=self.timeout, ) if obs_resp.status_code == 200: data = obs_resp.json() @@ -415,26 +445,47 @@ class WeatherDataCollector: # MGM 数据字段映射 # ruzgarHiz 实测为 km/h，转为 m/s 需要除以 3.6 ruz_hiz_kmh = latest.get("ruzgarHiz", 0) - + # MGM 返回 -9999 表示数据缺失，需要过滤 def _valid(v): return v is not None and v > -9000 - + results["current"] = { - "temp": latest.get("sicaklik") if _valid(latest.get("sicaklik")) else None, - "feels_like": latest.get("hissedilenSicaklik") if _valid(latest.get("hissedilenSicaklik")) else None, - "humidity": latest.get("nem") if _valid(latest.get("nem")) else None, - "wind_speed_ms": round(ruz_hiz_kmh / 3.6, 1) if _valid(ruz_hiz_kmh) else None, - "wind_speed_kt": round(ruz_hiz_kmh / 1.852, 1) if _valid(ruz_hiz_kmh) else None, - "wind_dir": latest.get("ruzgarYon") if _valid(latest.get("ruzgarYon")) else None, - "rain_24h": latest.get("toplamYagis") if _valid(latest.get("toplamYagis")) else None, - "pressure": latest.get("aktuelBasinc") if _valid(latest.get("aktuelBasinc")) else None, + "temp": latest.get("sicaklik") + if _valid(latest.get("sicaklik")) + else None, + "feels_like": latest.get("hissedilenSicaklik") + if _valid(latest.get("hissedilenSicaklik")) + else None, + "humidity": latest.get("nem") + if _valid(latest.get("nem")) + else None, + "wind_speed_ms": round(ruz_hiz_kmh / 3.6, 1) + if _valid(ruz_hiz_kmh) + else None, + "wind_speed_kt": round(ruz_hiz_kmh / 1.852, 1) + if _valid(ruz_hiz_kmh) + else None, + "wind_dir": latest.get("ruzgarYon") + if _valid(latest.get("ruzgarYon")) + else None, + "rain_24h": latest.get("toplamYagis") + if _valid(latest.get("toplamYagis")) + else None, + "pressure": latest.get("aktuelBasinc") + if _valid(latest.get("aktuelBasinc")) + else None, "cloud_cover": latest.get("kapalilik"), # 0-8 八分位云量 - "mgm_max_temp": latest.get("maxSicaklik") if _valid(latest.get("maxSicaklik")) else None, + "mgm_max_temp": latest.get("maxSicaklik") + if _valid(latest.get("maxSicaklik")) + else None, "time": latest.get("veriZamani"), - "station_name": latest.get("istasyonAd") or latest.get("adi") or latest.get("merkezAd") or "Ankara Esenboğa" + "station_name": latest.get("istasyonAd") + or latest.get("adi") + or latest.get("merkezAd") + or "Ankara Esenboğa", } - + # 2. 每日预报（尝试两个可能的 API 路径） forecast_urls = [ f"{base_url}/tahminler/gunluk?istno={istno}", @@ -442,7 +493,9 @@ class WeatherDataCollector: ] for forecast_url in forecast_urls: try: - daily_resp = self.session.get(forecast_url, headers=headers, timeout=self.timeout) + daily_resp = self.session.get( + forecast_url, headers=headers, timeout=self.timeout + ) if daily_resp.status_code == 200: forecasts = daily_resp.json() if forecasts and isinstance(forecasts, list): @@ -452,17 +505,29 @@ class WeatherDataCollector: if high_val is not None: results["today_high"] = high_val results["today_low"] = low_val - logger.info(f"📋 MGM 每日预报: 最高 {high_val}°C, 最低 {low_val}°C (from {forecast_url})") + logger.info( + f"📋 MGM 每日预报: 最高 {high_val}°C, 最低 {low_val}°C (from {forecast_url})" + ) break else: # 记录所有可用字段，方便调试 - available_keys = [k for k in today.keys() if "yuksek" in k.lower() or "sicaklik" in k.lower() or "gun" in k.lower()] - logger.warning(f"MGM 每日预报: enYuksekGun1 为空，可用字段: {available_keys}") + available_keys = [ + k + for k in today.keys() + if "yuksek" in k.lower() + or "sicaklik" in k.lower() + or "gun" in k.lower() + ] + logger.warning( + f"MGM 每日预报: enYuksekGun1 为空，可用字段: {available_keys}" + ) else: - logger.debug(f"MGM forecast URL {forecast_url} returned {daily_resp.status_code}") + logger.debug( + f"MGM forecast URL {forecast_url} returned {daily_resp.status_code}" + ) except Exception as e: logger.debug(f"MGM forecast URL {forecast_url} failed: {e}") - + return results if "current" in results else None except Exception as e: logger.error(f"MGM API 请求失败 ({istno}): {e}") @@ -477,24 +542,28 @@ class WeatherDataCollector: # 1. 获取网格点 points_url = f"https://api.weather.gov/points/{lat},{lon}" headers = {"User-Agent": "PolyWeather/1.0 (weather-bot)"} - - points_resp = self.session.get(points_url, headers=headers, timeout=self.timeout) + + points_resp = self.session.get( + points_url, headers=headers, timeout=self.timeout + ) points_resp.raise_for_status() points_data = points_resp.json() - + forecast_url = points_data.get("properties", {}).get("forecast") if not forecast_url: return None - + # 2. 获取预报 - forecast_resp = self.session.get(forecast_url, headers=headers, timeout=self.timeout) + forecast_resp = self.session.get( + forecast_url, headers=headers, timeout=self.timeout + ) forecast_resp.raise_for_status() forecast_data = forecast_resp.json() - + periods = forecast_data.get("properties", {}).get("periods", []) if not periods: return None - + # 3. 提取今日最高温（找 isDaytime=True 的第一个） today_high = None for p in periods: @@ -507,7 +576,7 @@ class WeatherDataCollector: if p.get("isDaytime"): today_high = p.get("temperature") break - + return { "source": "nws", "today_high": today_high, @@ -570,19 +639,19 @@ class WeatherDataCollector: if "temperature_2m_max_ecmwf_ifs04" in daily_data: ecmwf_max = daily_data.get("temperature_2m_max_ecmwf_ifs04", []) hrrr_max = daily_data.get("temperature_2m_max_ncep_hrrr_conus", []) - + # 记录今日模型分歧 daily_data["model_split"] = { "ecmwf": ecmwf_max[0] if ecmwf_max else None, - "hrrr": hrrr_max[0] if hrrr_max else None + "hrrr": hrrr_max[0] if hrrr_max else None, } - + # 智能合并：HRRR 仅覆盖 48 小时，远期用 ECMWF 补全 merged_max = [] for i in range(len(ecmwf_max)): hrrr_val = hrrr_max[i] if i < len(hrrr_max) else None ecmwf_val = ecmwf_max[i] if i < len(ecmwf_max) else None - + # 优先 HRRR，其次 ECMWF，都没有就跳过 if hrrr_val is not None: merged_max.append(hrrr_val) @@ -596,7 +665,9 @@ class WeatherDataCollector: # 映射逐小时数据 hourly_data = data.get("hourly", {}) if "temperature_2m_ncep_hrrr_conus" in hourly_data: - hourly_data["temperature_2m"] = hourly_data["temperature_2m_ncep_hrrr_conus"] + hourly_data["temperature_2m"] = hourly_data[ + "temperature_2m_ncep_hrrr_conus" + ] # 计算精确的当地时间 now_utc = datetime.utcnow() @@ -662,7 +733,7 @@ class WeatherDataCollector: if key.startswith("temperature_2m_max") and key != "temperature_2m_max": if values and values[0] is not None: today_highs.append(values[0]) - + # 也检查非成员键（有些返回格式不同） if not today_highs: raw_max = daily.get("temperature_2m_max", []) @@ -712,14 +783,14 @@ class WeatherDataCollector: """ 从 Open-Meteo 获取多个独立 NWP 模型的预报 用于真正的多模型共识评分 - + 模型列表: - ECMWF IFS (欧洲中期天气预报中心) - GFS (美国 NOAA) - ICON (德国气象局 DWD) - GEM (加拿大气象局) - JMA (日本气象厅) - + 返回 3 天的预报数据，支持今日+明日共识分析 """ try: @@ -748,7 +819,7 @@ class WeatherDataCollector: daily = data.get("daily", {}) dates = daily.get("time", []) - + model_labels = { "ecmwf_ifs025": "ECMWF", "gfs_seamless": "GFS", @@ -756,7 +827,7 @@ class WeatherDataCollector: "gem_seamless": "GEM", "jma_seamless": "JMA", } - + # 按天提取每个模型的预报 daily_forecasts = {} # {"2026-02-23": {"ECMWF": 7.9, "GFS": 6.5, ...}, ...} for day_idx, date_str in enumerate(dates): @@ -778,8 +849,10 @@ class WeatherDataCollector: forecasts = daily_forecasts.get(today_date, {}) labels_str = ", ".join([f"{k}={v}" for k, v in forecasts.items()]) - logger.info(f"🔬 Multi-model ({len(forecasts)}个, {len(daily_forecasts)}天): {labels_str}") - + logger.info( + f"🔬 Multi-model ({len(forecasts)}个, {len(daily_forecasts)}天): {labels_str}" + ) + return { "source": "multi_model", "forecasts": forecasts, # 今天 {"ECMWF": 12.3, "GFS": 11.8, ...} (向后兼容) @@ -814,47 +887,47 @@ class WeatherDataCollector: "lat": lat, "lon": lon, "format": "json", - "as_daylight": "true" + "as_daylight": "true", } - - response = self.session.get( - url, - params=params, - timeout=self.timeout - ) + + response = self.session.get(url, params=params, timeout=self.timeout) response.raise_for_status() data = response.json() - + day_data = data.get("data_day", {}) max_temps = day_data.get("temperature_max", []) - + if not max_temps: - logger.warning(f"Meteoblue API 返回数据中找不到最高温 (坐标: {lat},{lon})") + logger.warning( + f"Meteoblue API 返回数据中找不到最高温 (坐标: {lat},{lon})" + ) return None # 2. 转换单位 def c_to_f(c): - return round((c * 9/5) + 32, 1) + return round((c * 9 / 5) + 32, 1) result = { "source": "meteoblue", "today_high": None, "daily_highs": [], "unit": "fahrenheit" if use_fahrenheit else "celsius", - "url": f"https://www.meteoblue.com/en/weather/week/{lat}N{lon}E" # 仅供参考 + "url": f"https://www.meteoblue.com/en/weather/week/{lat}N{lon}E", # 仅供参考 } # 提取今日最高 mb_today_c = max_temps[0] result["today_high"] = c_to_f(mb_today_c) if use_fahrenheit else mb_today_c - + # 提取接下来几天的最高温 if use_fahrenheit: result["daily_highs"] = [c_to_f(t) for t in max_temps] else: result["daily_highs"] = max_temps - logger.info(f"✅ Meteoblue API 获取成功 ({lat},{lon}): 今天 {result['today_high']}{result['unit']}") + logger.info( + f"✅ Meteoblue API 获取成功 ({lat},{lon}): 今天 {result['today_high']}{result['unit']}" + ) return result except Exception as e: logger.error(f"Meteoblue API fetch failed: {e}") @@ -867,9 +940,18 @@ class WeatherDataCollector: """ # 1. 尝试英文月份 months = { - "January": "01", "February": "02", "March": "03", "April": "04", - "May": "05", "June": "06", "July": "07", "August": "08", - "September": "09", "October": "10", "November": "11", "December": "12", + "January": "01", + "February": "02", + "March": "03", + "April": "04", + "May": "05", + "June": "06", + "July": "07", + "August": "08", + "September": "09", + "October": "10", + "November": "11", + "December": "12", } for month_name, month_val in months.items(): if month_name in title: @@ -886,7 +968,7 @@ class WeatherDataCollector: day = int(zh_match.group(2)) year = datetime.now().year return f"{year}-{month:02d}-{day:02d}" - + # 3. 尝试 ISO 格式 YYYY-MM-DD iso_match = re.search(r"(\d{4})-(\d{2})-(\d{2})", title) if iso_match: @@ -900,21 +982,21 @@ class WeatherDataCollector: """ # 坐标使用 METAR 机场位置（Polymarket 以机场数据结算） static_coords = { - "london": {"lat": 51.5053, "lon": 0.0553}, # EGLC London City - "paris": {"lat": 49.0097, "lon": 2.5478}, # LFPG Charles de Gaulle - "new york": {"lat": 40.7750, "lon": -73.8750}, # KLGA LaGuardia + "london": {"lat": 51.5053, "lon": 0.0553}, # EGLC London City + "paris": {"lat": 49.0097, "lon": 2.5478}, # LFPG Charles de Gaulle + "new york": {"lat": 40.7750, "lon": -73.8750}, # KLGA LaGuardia "new york's central park": {"lat": 40.7812, "lon": -73.9665}, - "nyc": {"lat": 40.7750, "lon": -73.8750}, # KLGA LaGuardia - "seattle": {"lat": 47.4499, "lon": -122.3118}, # KSEA Sea-Tac - "chicago": {"lat": 41.9769, "lon": -87.9081}, # KORD O'Hare - "dallas": {"lat": 32.8459, "lon": -96.8509}, # KDAL Love Field - "miami": {"lat": 25.7933, "lon": -80.2906}, # KMIA International - "atlanta": {"lat": 33.6367, "lon": -84.4281}, # KATL Hartsfield-Jackson - "seoul": {"lat": 37.4691, "lon": 126.4510}, # RKSI Incheon - "toronto": {"lat": 43.6759, "lon": -79.6294}, # CYYZ Pearson - "ankara": {"lat": 40.1281, "lon": 32.9950}, # LTAC Esenboğa - "wellington": {"lat": -41.3272, "lon": 174.8053}, # NZWN Wellington - "buenos aires": {"lat": -34.8222, "lon": -58.5358}, # SAEZ Ezeiza + "nyc": {"lat": 40.7750, "lon": -73.8750}, # KLGA LaGuardia + "seattle": {"lat": 47.4499, "lon": -122.3118}, # KSEA Sea-Tac + "chicago": {"lat": 41.9769, "lon": -87.9081}, # KORD O'Hare + "dallas": {"lat": 32.8459, "lon": -96.8509}, # KDAL Love Field + "miami": {"lat": 25.7933, "lon": -80.2906}, # KMIA International + "atlanta": {"lat": 33.6367, "lon": -84.4281}, # KATL Hartsfield-Jackson + "seoul": {"lat": 37.4691, "lon": 126.4510}, # RKSI Incheon + "toronto": {"lat": 43.6759, "lon": -79.6294}, # CYYZ Pearson + "ankara": {"lat": 40.1281, "lon": 32.9950}, # LTAC Esenboğa + "wellington": {"lat": -41.3272, "lon": 174.8053}, # NZWN Wellington + "buenos aires": {"lat": -34.8222, "lon": -58.5358}, # SAEZ Ezeiza } normalized_city = city.lower().strip() @@ -956,30 +1038,64 @@ class WeatherDataCollector: # 1. 优先尝试已知城市列表 (硬编码匹配) known_cities = { - "london": "London", "伦敦": "London", - "new york": "New York", "new york's central park": "New York", "nyc": "New York", "纽约": "New York", - "seattle": "Seattle", "西雅图": "Seattle", - "chicago": "Chicago", "芝加哥": "Chicago", - "dallas": "Dallas", "达拉斯": "Dallas", - "miami": "Miami", "迈阿密": "Miami", - "atlanta": "Atlanta", "亚特兰大": "Atlanta", - "seoul": "Seoul", "首尔": "Seoul", - "toronto": "Toronto", "多伦多": "Toronto", - "ankara": "Ankara", "安卡拉": "Ankara", - "wellington": "Wellington", "惠灵顿": "Wellington", - "buenos aires": "Buenos Aires", "布宜诺斯艾利斯": "Buenos Aires" + "london": "London", + "伦敦": "London", + "new york": "New York", + "new york's central park": "New York", + "nyc": "New York", + "纽约": "New York", + "seattle": "Seattle", + "西雅图": "Seattle", + "chicago": "Chicago", + "芝加哥": "Chicago", + "dallas": "Dallas", + "达拉斯": "Dallas", + "miami": "Miami", + "迈阿密": "Miami", + "atlanta": "Atlanta", + "亚特兰大": "Atlanta", + "seoul": "Seoul", + "首尔": "Seoul", + "toronto": "Toronto", + "多伦多": "Toronto", + "ankara": "Ankara", + "安卡拉": "Ankara", + "wellington": "Wellington", + "惠灵顿": "Wellington", + "buenos aires": "Buenos Aires", + "布宜诺斯艾利斯": "Buenos Aires", } - + for key, val in known_cities.items(): if key in q: return val # 2. 从英文模板中提取 - triggers = ["temperature in ", "temp in ", "weather in ", "highest-temperature-in-", "temperature-in-"] + triggers = [ + "temperature in ", + "temp in ", + "weather in ", + "highest-temperature-in-", + "temperature-in-", + ] for trigger in triggers: if trigger in q: part = q.split(trigger)[1] - delimiters = [" on ", " at ", " above ", " below ", " be ", " is ", " will ", " has ", " reached ", "?", " (", ", ", "-"] + delimiters = [ + " on ", + " at ", + " above ", + " below ", + " be ", + " is ", + " will ", + " has ", + " reached ", + "?", + " (", + ", ", + "-", + ] city = part for d in delimiters: if d in city: @@ -1039,22 +1155,25 @@ class WeatherDataCollector: results["open-meteo"] = open_meteo # 获取时区偏移以过滤 METAR utc_offset = open_meteo.get("utc_offset", 0) - metar_data = self.fetch_metar(city, use_fahrenheit=use_fahrenheit, utc_offset=utc_offset) + metar_data = self.fetch_metar( + city, use_fahrenheit=use_fahrenheit, utc_offset=utc_offset + ) if metar_data: results["metar"] = metar_data - + # 对安卡拉，额外获取 MGM 官方数据 if city_lower == "ankara": mgm_data = self.fetch_from_mgm("17128") if mgm_data: results["mgm"] = mgm_data - + # 对伦敦，获取 Meteoblue 预测 (公认最准) if city_lower == "london": mb_data = self.fetch_from_meteoblue( - lat, lon, + lat, + lon, timezone_name=open_meteo.get("timezone", "UTC"), - use_fahrenheit=use_fahrenheit + use_fahrenheit=use_fahrenheit, ) if mb_data: results["meteoblue"] = mb_data @@ -1064,14 +1183,16 @@ class WeatherDataCollector: nws_data = self.fetch_nws(lat, lon) if nws_data: results["nws"] = nws_data - + # 集合预报 (所有城市通用，用于不确定性分析) ens_data = self.fetch_ensemble(lat, lon, use_fahrenheit=use_fahrenheit) if ens_data: results["ensemble"] = ens_data - + # 多模型预报 (所有城市通用，用于共识评分) - mm_data = self.fetch_multi_model(lat, lon, use_fahrenheit=use_fahrenheit) + mm_data = self.fetch_multi_model( + lat, lon, use_fahrenheit=use_fahrenheit + ) if mm_data: results["multi_model"] = mm_data else: diff --git a/src/data_mining/fetch_history.py b/src/data_mining/fetch_history.py index 64e7f0b8..643168f5 100644 --- a/src/data_mining/fetch_history.py +++ b/src/data_mining/fetch_history.py @@ -1,27 +1,28 @@ 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') +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'] - + 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}" @@ -29,57 +30,68 @@ def fetch_historical_data_for_city(city_info, output_dir): "cloud_cover,shortwave_radiation,precipitation,surface_pressure" "&timezone=auto" ) - - logging.info(f"Downloading historical data for {city_name} (Lat: {lat}, Lon: {lon})...") - + + 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") + 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}") - + + 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') - + 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: + + 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', []) + + 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() diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py index f0424c53..301bbfb7 100644 --- a/src/utils/config_loader.py +++ b/src/utils/config_loader.py @@ -1,12 +1,13 @@ import os from dotenv import load_dotenv + def load_config(): """ Load configuration from environment variables and config files """ load_dotenv() - + def get_env_or_none(key): val = os.getenv(key) if not val or "your_" in val.lower() or val.strip() == "": @@ -40,14 +41,14 @@ def load_config(): "market_volume_signal": 0.15, "orderbook_analysis": 0.10, "technical_indicators": 0.05, - "onchain_whale_signal": 0.05 + "onchain_whale_signal": 0.05, } }, "app": { "log_level": os.getenv("LOG_LEVEL", "INFO"), "env": os.getenv("ENV", "development"), "proxy": os.getenv("HTTPS_PROXY") or os.getenv("HTTP_PROXY"), - } + }, } - + return config diff --git a/src/utils/logger.py b/src/utils/logger.py index ee315620..6b03339e 100644 --- a/src/utils/logger.py +++ b/src/utils/logger.py @@ -1,19 +1,20 @@ import sys from loguru import logger + def setup_logger(level="DEBUG"): """ Configure loguru logger """ logger.remove() # Remove default handler - + # 控制台输出 - 使用支持中文的格式 logger.add( sys.stderr, format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {message}", - level=level + level=level, ) - + # 文件输出 logger.add( "data/logs/trading_system.log", @@ -21,7 +22,7 @@ def setup_logger(level="DEBUG"): retention="10 days", level=level, encoding="utf-8", - compression="zip" + compression="zip", ) logger.info("日志系统初始化完成。")