From ffb88ae61d05936c0d64e15522f77bfc16c2b565 Mon Sep 17 00:00:00 2001
From: "2569718930@qq.com" <2569718930@qq.com>
Date: Wed, 11 Mar 2026 03:40:25 +0800
Subject: [PATCH] feat: implement multi-source weather data collection
including OpenWeatherMap, Visual Crossing, and METAR with caching.
---
bot_listener.py | 34 +++++-
src/data_collection/weather_sources.py | 138 ++++++++++++++++++++++---
2 files changed, 157 insertions(+), 15 deletions(-)
diff --git a/bot_listener.py b/bot_listener.py
index 0577c9be..88852708 100644
--- a/bot_listener.py
+++ b/bot_listener.py
@@ -392,6 +392,14 @@ def start_bot():
# --- 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"
+ if time_str == "N/A":
+ metar_obs = metar.get("observation_time", "") if metar else ""
+ if "T" in metar_obs:
+ time_str = metar_obs.split("T")[1][:5]
+ elif " " in metar_obs:
+ time_str = metar_obs.split(" ")[1][:5]
+ elif metar_obs:
+ time_str = str(metar_obs)[:5]
risk_profile = get_city_risk_profile(city_name)
risk_emoji = risk_profile.get("risk_level", "⚠️") if risk_profile else "⚠️"
@@ -414,11 +422,27 @@ def start_bot():
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"))
+ metar_max_so_far = _sf(metar.get("current", {}).get("max_temp_so_far")) if metar else None
# 今天对比
- today_t = max_temps[0] if max_temps else "N/A"
+ today_t = _sf(max_temps[0]) if max_temps else None
+ fallback_source = None
+ if today_t is None:
+ for source_name, candidate in (
+ ("MB", mb_high),
+ ("NWS", nws_high),
+ ("MGM", mgm_high),
+ ("METAR", metar_max_so_far),
+ ):
+ if candidate is not None:
+ today_t = candidate
+ fallback_source = source_name
+ break
+ today_t_display = (
+ f"{today_t:.1f}" if isinstance(today_t, (int, float)) else "N/A"
+ )
comp_parts = []
- sources = ["Open-Meteo"]
+ sources = ["Open-Meteo"] if max_temps else []
if mb_high is not None:
sources.append("MB")
@@ -441,6 +465,10 @@ def start_bot():
if isinstance(mgm_high, (int, float))
else f"🇺🇸 MGM: {mgm_high}"
)
+ if fallback_source and fallback_source not in sources:
+ sources.append(fallback_source)
+ if not sources:
+ sources = ["N/A"]
# 检查是否有显著分歧 (超过 5°F 或 2.5°C)
divergence_warning = ""
@@ -457,7 +485,7 @@ def start_bot():
msg_lines.append(f"\n📊 预报 ({sources_str})")
msg_lines.append(
- f"👉 今天: {today_t}{temp_symbol}{comp_str}{divergence_warning}"
+ f"👉 今天: {today_t_display}{temp_symbol}{comp_str}{divergence_warning}"
)
# 明后天
diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py
index 74969853..5ead8b02 100644
--- a/src/data_collection/weather_sources.py
+++ b/src/data_collection/weather_sources.py
@@ -64,6 +64,21 @@ class WeatherDataCollector:
self.timeout = 30 # 增加超时以支持高延迟 VPS
self.session = requests.Session()
+ self.open_meteo_cache_ttl_sec = int(
+ os.getenv("OPEN_METEO_CACHE_TTL_SEC", "900")
+ )
+ self.open_meteo_ensemble_cache_ttl_sec = int(
+ os.getenv("OPEN_METEO_ENSEMBLE_CACHE_TTL_SEC", "900")
+ )
+ self.open_meteo_multi_model_cache_ttl_sec = int(
+ os.getenv("OPEN_METEO_MULTI_MODEL_CACHE_TTL_SEC", "900")
+ )
+ self._open_meteo_cache: Dict[str, Dict] = {}
+ self._ensemble_cache: Dict[str, Dict] = {}
+ self._multi_model_cache: Dict[str, Dict] = {}
+ self._open_meteo_cache_lock = threading.Lock()
+ self._ensemble_cache_lock = threading.Lock()
+ self._multi_model_cache_lock = threading.Lock()
self.meteoblue_cache_ttl_sec = int(
os.getenv("METEOBLUE_CACHE_TTL_SEC", "1800")
)
@@ -265,7 +280,6 @@ class WeatherDataCollector:
response = self.session.get(
url,
params=params,
- headers={"Cache-Control": "no-cache", "Pragma": "no-cache"},
timeout=self.timeout,
)
response.raise_for_status()
@@ -931,6 +945,20 @@ class WeatherDataCollector:
forecast_days: Number of forecast days to fetch (default 14 to cover all market dates)
use_fahrenheit: Whether to return temperatures in Fahrenheit (for US markets)
"""
+ cache_key = (
+ f"{round(float(lat), 4)}:{round(float(lon), 4)}:"
+ f"{forecast_days}:{'f' if use_fahrenheit else 'c'}"
+ )
+ now_ts = time.time()
+ with self._open_meteo_cache_lock:
+ cached = self._open_meteo_cache.get(cache_key)
+ if (
+ cached
+ and now_ts - float(cached.get("t", 0)) < self.open_meteo_cache_ttl_sec
+ ):
+ cached_data = cached.get("data")
+ if isinstance(cached_data, dict):
+ return dict(cached_data)
try:
url = "https://api.open-meteo.com/v1/forecast"
params = {
@@ -941,7 +969,6 @@ class WeatherDataCollector:
"daily": "temperature_2m_max,apparent_temperature_max,sunrise,sunset,sunshine_duration",
"timezone": "auto",
"forecast_days": forecast_days,
- "_t": int(time.time()), # 禁用缓存，强制刷新
}
# 显式指定单位，防止 API 默认行为漂移
@@ -953,7 +980,6 @@ class WeatherDataCollector:
response = self.session.get(
url,
params=params,
- headers={"Cache-Control": "no-cache", "Pragma": "no-cache"},
timeout=self.timeout,
)
response.raise_for_status()
@@ -1003,7 +1029,7 @@ class WeatherDataCollector:
local_now = now_utc + timedelta(seconds=utc_offset)
local_time_str = local_now.strftime("%Y-%m-%d %H:%M")
- return {
+ result = {
"source": "open-meteo",
"timestamp": now_utc.isoformat(),
"timezone": timezone_name,
@@ -1016,8 +1042,26 @@ class WeatherDataCollector:
"daily": daily_data,
"unit": "fahrenheit" if use_fahrenheit else "celsius",
}
+ with self._open_meteo_cache_lock:
+ self._open_meteo_cache[cache_key] = {
+ "t": time.time(),
+ "data": dict(result),
+ }
+ return result
except Exception as e:
- logger.error(f"Open-Meteo forecast failed: {e}")
+ status_code = getattr(getattr(e, "response", None), "status_code", None)
+ if status_code == 429:
+ logger.warning(
+ f"Open-Meteo rate limited (429), fallback to cache if available: lat={lat}, lon={lon}"
+ )
+ else:
+ logger.error(f"Open-Meteo forecast failed: {e}")
+ with self._open_meteo_cache_lock:
+ stale = self._open_meteo_cache.get(cache_key)
+ if stale and isinstance(stale.get("data"), dict):
+ fallback = dict(stale["data"])
+ fallback["stale_cache"] = True
+ return fallback
return None
def fetch_ensemble(
@@ -1030,6 +1074,21 @@ class WeatherDataCollector:
从 Open-Meteo Ensemble API 获取 51 成员集合预报
用于计算预报不确定性范围（散度）
"""
+ cache_key = (
+ f"{round(float(lat), 4)}:{round(float(lon), 4)}:"
+ f"{'f' if use_fahrenheit else 'c'}"
+ )
+ now_ts = time.time()
+ with self._ensemble_cache_lock:
+ cached = self._ensemble_cache.get(cache_key)
+ if (
+ cached
+ and now_ts - float(cached.get("t", 0))
+ < self.open_meteo_ensemble_cache_ttl_sec
+ ):
+ cached_data = cached.get("data")
+ if isinstance(cached_data, dict):
+ return dict(cached_data)
try:
url = "https://ensemble-api.open-meteo.com/v1/ensemble"
params = {
@@ -1038,7 +1097,6 @@ class WeatherDataCollector:
"daily": "temperature_2m_max",
"timezone": "auto",
"forecast_days": 3,
- "_t": int(time.time()),
}
if use_fahrenheit:
params["temperature_unit"] = "fahrenheit"
@@ -1048,7 +1106,6 @@ class WeatherDataCollector:
response = self.session.get(
url,
params=params,
- headers={"Cache-Control": "no-cache"},
timeout=self.timeout,
)
response.raise_for_status()
@@ -1098,9 +1155,26 @@ class WeatherDataCollector:
f"📊 Ensemble ({n} members): median={median:.1f}, "
f"p10={p10:.1f}, p90={p90:.1f}"
)
+ with self._ensemble_cache_lock:
+ self._ensemble_cache[cache_key] = {
+ "t": time.time(),
+ "data": dict(result),
+ }
return result
except Exception as e:
- logger.warning(f"Ensemble API 请求失败: {e}")
+ status_code = getattr(getattr(e, "response", None), "status_code", None)
+ if status_code == 429:
+ logger.warning(
+ f"Ensemble API rate limited (429), fallback to cache if available: lat={lat}, lon={lon}"
+ )
+ else:
+ logger.warning(f"Ensemble API 请求失败: {e}")
+ with self._ensemble_cache_lock:
+ stale = self._ensemble_cache.get(cache_key)
+ if stale and isinstance(stale.get("data"), dict):
+ fallback = dict(stale["data"])
+ fallback["stale_cache"] = True
+ return fallback
return None
def fetch_multi_model(
@@ -1122,6 +1196,21 @@ class WeatherDataCollector:
返回 3 天的预报数据，支持今日+明日共识分析
"""
+ cache_key = (
+ f"{round(float(lat), 4)}:{round(float(lon), 4)}:"
+ f"{'f' if use_fahrenheit else 'c'}"
+ )
+ now_ts = time.time()
+ with self._multi_model_cache_lock:
+ cached = self._multi_model_cache.get(cache_key)
+ if (
+ cached
+ and now_ts - float(cached.get("t", 0))
+ < self.open_meteo_multi_model_cache_ttl_sec
+ ):
+ cached_data = cached.get("data")
+ if isinstance(cached_data, dict):
+ return dict(cached_data)
try:
url = "https://api.open-meteo.com/v1/forecast"
models = "ecmwf_ifs025,gfs_seamless,icon_seamless,gem_seamless,jma_seamless"
@@ -1132,7 +1221,6 @@ class WeatherDataCollector:
"models": models,
"timezone": "auto",
"forecast_days": 3,
- "_t": int(time.time()),
}
if use_fahrenheit:
params["temperature_unit"] = "fahrenheit"
@@ -1140,7 +1228,6 @@ class WeatherDataCollector:
response = self.session.get(
url,
params=params,
- headers={"Cache-Control": "no-cache"},
timeout=self.timeout,
)
response.raise_for_status()
@@ -1182,15 +1269,33 @@ class WeatherDataCollector:
f"🔬 Multi-model ({len(forecasts)}个, {len(daily_forecasts)}天): {labels_str}"
)
- return {
+ result = {
"source": "multi_model",
"forecasts": forecasts, # 今天 {"ECMWF": 12.3, "GFS": 11.8, ...} (向后兼容)
"daily_forecasts": daily_forecasts, # 按天 {"2026-02-23": {...}, "2026-02-24": {...}}
"dates": dates,
"unit": "fahrenheit" if use_fahrenheit else "celsius",
}
+ with self._multi_model_cache_lock:
+ self._multi_model_cache[cache_key] = {
+ "t": time.time(),
+ "data": dict(result),
+ }
+ return result
except Exception as e:
- logger.warning(f"Multi-model API 请求失败: {e}")
+ status_code = getattr(getattr(e, "response", None), "status_code", None)
+ if status_code == 429:
+ logger.warning(
+ f"Multi-model API rate limited (429), fallback to cache if available: lat={lat}, lon={lon}"
+ )
+ else:
+ logger.warning(f"Multi-model API 请求失败: {e}")
+ with self._multi_model_cache_lock:
+ stale = self._multi_model_cache.get(cache_key)
+ if stale and isinstance(stale.get("data"), dict):
+ fallback = dict(stale["data"])
+ fallback["stale_cache"] = True
+ return fallback
return None
def fetch_from_meteoblue(
@@ -1586,6 +1691,15 @@ class WeatherDataCollector:
metar_data = self.fetch_metar(city, use_fahrenheit=use_fahrenheit)
if metar_data:
results["metar"] = metar_data
+ if city_lower in self.METEOBLUE_PRIORITY_CITIES:
+ mb_data = self.fetch_from_meteoblue(
+ lat,
+ lon,
+ timezone_name="UTC",
+ use_fahrenheit=use_fahrenheit,
+ )
+ if mb_data:
+ results["meteoblue"] = mb_data
if use_fahrenheit:
nws_data = self.fetch_nws(lat, lon)
if nws_data: