From 8712335ffd89cb7aaf324bba6f39869401f9e0be Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 11 Mar 2026 05:42:05 +0800 Subject: [PATCH] feat: Implement PolyWeather web map API by reusing existing analysis logic and centralize weather data collection. --- bot_listener.py | 45 +++++++++++- src/data_collection/weather_sources.py | 97 ++++++++++++++++++++------ web/app.py | 59 +++++++++++++++- 3 files changed, 174 insertions(+), 27 deletions(-) diff --git a/bot_listener.py b/bot_listener.py index 54a094d7..7b3c83a7 100644 --- a/bot_listener.py +++ b/bot_listener.py @@ -14,6 +14,7 @@ from src.utils.telegram_push import start_trade_alert_push_loop # type: ignore from src.onchain.polygon_wallet_watcher import start_polygon_wallet_watch_loop # type: ignore # noqa: E402 from src.onchain.polymarket_wallet_activity_watcher import start_polymarket_wallet_activity_loop # type: ignore # noqa: E402 from src.data_collection.weather_sources import WeatherDataCollector # type: ignore # noqa: E402 +from src.data_collection.city_registry import CITY_REGISTRY # 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 from src.database.db_manager import DBManager @@ -376,6 +377,20 @@ def start_bot(): open_meteo = weather_data.get("open-meteo", {}) metar = weather_data.get("metar", {}) mgm = weather_data.get("mgm") or {} + city_meta = CITY_REGISTRY.get(city_name.lower(), {}) + fallback_utc_offset = int(city_meta.get("tz_offset", 0)) + nws_periods = (weather_data.get("nws", {}) or {}).get("forecast_periods", []) or [] + if nws_periods: + try: + from datetime import datetime as _dt + + first_start = nws_periods[0].get("start_time") + if first_start: + maybe_dt = _dt.fromisoformat(str(first_start)) + if maybe_dt.utcoffset() is not None: + fallback_utc_offset = int(maybe_dt.utcoffset().total_seconds()) + except Exception: + pass # 鏁板€煎綊涓€鍖? def _sf(v): @@ -395,11 +410,33 @@ def start_bot(): 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] + try: + from datetime import datetime, timezone, timedelta + + dt = datetime.fromisoformat(metar_obs.replace("Z", "+00:00")) + utc_offset_for_view = open_meteo.get("utc_offset") + if utc_offset_for_view is None: + utc_offset_for_view = fallback_utc_offset + local_dt = dt.astimezone( + timezone(timedelta(seconds=int(utc_offset_for_view))) + ) + time_str = local_dt.strftime("%H:%M") + except Exception: + 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] + else: + try: + from datetime import datetime, timezone, timedelta + + local_now = datetime.now(timezone.utc).astimezone( + timezone(timedelta(seconds=int(fallback_utc_offset))) + ) + time_str = local_now.strftime("%H:%M") + except Exception: + pass risk_profile = get_city_risk_profile(city_name) risk_emoji = risk_profile.get("risk_level", "⚠️") if risk_profile else "⚠️" @@ -556,9 +593,11 @@ def start_bot(): from datetime import datetime, timezone, timedelta dt = datetime.fromisoformat(obs_t.replace("Z", "+00:00")) - utc_offset = open_meteo.get("utc_offset", 0) + utc_offset = open_meteo.get("utc_offset") + if utc_offset is None: + utc_offset = fallback_utc_offset local_dt = dt.astimezone( - timezone(timedelta(seconds=utc_offset)) + timezone(timedelta(seconds=int(utc_offset))) ) obs_t_str = local_dt.strftime("%H:%M") # 计算数据年龄 diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index dcb57156..1e63569b 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -98,10 +98,17 @@ class WeatherDataCollector: # 磁盘持久化缓存：重启后即可加载上次的预报数据，避免冷启动请求爆发 self._disk_cache_path = os.getenv( - "OPEN_METEO_DISK_CACHE_PATH", "/app/.cache/open_meteo_cache.json" + "OPEN_METEO_DISK_CACHE_PATH", "/app/data/open_meteo_cache.json" + ) + self._disk_cache_max_age_sec = int( + os.getenv("OPEN_METEO_DISK_CACHE_MAX_AGE_SEC", "86400") ) self._disk_cache_lock = threading.Lock() + self._disk_cache_last_mtime: float = 0.0 self._load_open_meteo_disk_cache() + logger.info( + f"Open-Meteo 磁盘缓存路径: {self._disk_cache_path} (max_age={self._disk_cache_max_age_sec}s)" + ) # 设置代理 proxy = config.get("proxy") @@ -117,35 +124,70 @@ class WeatherDataCollector: """启动时从磁盘加载 Open-Meteo 三类缓存，避免重启后冷启动打爆 API""" import json as _json try: - if not os.path.exists(self._disk_cache_path): + path = self._disk_cache_path + if not os.path.exists(path): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + _json.dump( + { + "forecast": {}, + "ensemble": {}, + "multi_model": {}, + "saved_at": time.time(), + }, + f, + ) + self._disk_cache_last_mtime = os.path.getmtime(path) return - with open(self._disk_cache_path, "r", encoding="utf-8") as f: + current_mtime = os.path.getmtime(path) + if current_mtime <= self._disk_cache_last_mtime: + return + with open(path, "r", encoding="utf-8") as f: saved = _json.load(f) now = time.time() - # 最多允许加载 2× TTL 的旧数据（TTL 内视为新鲜，超出则过期但仍可作为 stale fallback） - max_age = max( - self.open_meteo_cache_ttl_sec, - self.open_meteo_ensemble_cache_ttl_sec, - self.open_meteo_multi_model_cache_ttl_sec, - ) * 2 + max_age = max(600, self._disk_cache_max_age_sec) loaded = 0 - for key, entry in saved.get("forecast", {}).items(): - if now - float(entry.get("t", 0)) < max_age: - self._open_meteo_cache[key] = entry - loaded += 1 - for key, entry in saved.get("ensemble", {}).items(): - if now - float(entry.get("t", 0)) < max_age: - self._ensemble_cache[key] = entry - loaded += 1 - for key, entry in saved.get("multi_model", {}).items(): - if now - float(entry.get("t", 0)) < max_age: - self._multi_model_cache[key] = entry - loaded += 1 + with self._open_meteo_cache_lock: + for key, entry in saved.get("forecast", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + old = self._open_meteo_cache.get(key) + if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)): + self._open_meteo_cache[key] = entry + loaded += 1 + with self._ensemble_cache_lock: + for key, entry in saved.get("ensemble", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + old = self._ensemble_cache.get(key) + if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)): + self._ensemble_cache[key] = entry + loaded += 1 + with self._multi_model_cache_lock: + for key, entry in saved.get("multi_model", {}).items(): + if now - float(entry.get("t", 0)) < max_age: + old = self._multi_model_cache.get(key) + if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)): + self._multi_model_cache[key] = entry + loaded += 1 + self._disk_cache_last_mtime = current_mtime if loaded: logger.info(f"✅ 从磁盘加载 Open-Meteo 缓存 {loaded} 条 ({self._disk_cache_path})") except Exception as e: logger.warning(f"磁盘缓存加载失败（首次启动不影响运行）: {e}") + def _maybe_reload_open_meteo_disk_cache(self) -> None: + """跨进程共享缓存：当缓存文件有更新时增量重载到当前进程内存""" + try: + path = self._disk_cache_path + if not os.path.exists(path): + return + current_mtime = os.path.getmtime(path) + if current_mtime <= self._disk_cache_last_mtime: + return + self._load_open_meteo_disk_cache() + except Exception: + # 不影响主流程 + pass + def _flush_open_meteo_disk_cache(self) -> None: """将三类 Open-Meteo 内存缓存持久化到磁盘""" import json as _json @@ -168,6 +210,7 @@ class WeatherDataCollector: with open(tmp_path, "w", encoding="utf-8") as f: _json.dump(payload, f) os.replace(tmp_path, self._disk_cache_path) # 原子替换，防止写入一半时被读到 + self._disk_cache_last_mtime = os.path.getmtime(self._disk_cache_path) except Exception as e: logger.warning(f"磁盘缓存写入失败: {e}") @@ -1041,6 +1084,7 @@ class WeatherDataCollector: f"{round(float(lat), 4)}:{round(float(lon), 4)}:" f"{forecast_days}:{'f' if use_fahrenheit else 'c'}" ) + self._maybe_reload_open_meteo_disk_cache() now_ts = time.time() # ── 429 冷却期检查（所有 Open-Meteo 端点共享）───────────────── with self._open_meteo_rl_lock: @@ -1195,6 +1239,7 @@ class WeatherDataCollector: f"{round(float(lat), 4)}:{round(float(lon), 4)}:" f"{'f' if use_fahrenheit else 'c'}" ) + self._maybe_reload_open_meteo_disk_cache() now_ts = time.time() # ── 429 冷却期检查（所有 Open-Meteo 端点共享）───────────────── with self._open_meteo_rl_lock: @@ -1340,6 +1385,7 @@ class WeatherDataCollector: f"{round(float(lat), 4)}:{round(float(lon), 4)}:" f"{'f' if use_fahrenheit else 'c'}" ) + self._maybe_reload_open_meteo_disk_cache() now_ts = time.time() # ── 429 冷却期检查（所有 Open-Meteo 端点共享）───────────────── with self._open_meteo_rl_lock: @@ -1851,7 +1897,14 @@ class WeatherDataCollector: results["multi_model"] = mm_data else: # Open-Meteo 失败时，仍然尝试获取 METAR 和 NWS - metar_data = self.fetch_metar(city, use_fahrenheit=use_fahrenheit) + fallback_utc_offset = int( + self.CITY_REGISTRY.get(city_lower, {}).get("tz_offset", 0) + ) + metar_data = self.fetch_metar( + city, + use_fahrenheit=use_fahrenheit, + utc_offset=fallback_utc_offset, + ) if metar_data: results["metar"] = metar_data if city_lower in self.METEOBLUE_PRIORITY_CITIES: diff --git a/web/app.py b/web/app.py index 8618f227..1e3d551f 100644 --- a/web/app.py +++ b/web/app.py @@ -138,8 +138,21 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: obs_time_str = "" metar_age_min = None obs_t = metar.get("observation_time", "") if metar else "" - # 优先从 API 获取偏移，若失败则使用 CITIES 预设的静态偏移 (兜底当地时间) - utc_offset = om.get("utc_offset", info.get("tz", 0)) + # 优先从 API 获取偏移；若缺失则尝试 NWS 动态偏移；最后回退静态配置 + utc_offset = om.get("utc_offset") + if utc_offset is None: + try: + nws_periods = (raw.get("nws", {}) or {}).get("forecast_periods", []) or [] + if nws_periods: + first_start = nws_periods[0].get("start_time") + if first_start: + maybe_dt = datetime.fromisoformat(str(first_start)) + if maybe_dt.utcoffset() is not None: + utc_offset = int(maybe_dt.utcoffset().total_seconds()) + except Exception: + utc_offset = None + if utc_offset is None: + utc_offset = info.get("tz", 0) if obs_t and "T" in obs_t: try: dt = datetime.fromisoformat(obs_t.replace("Z", "+00:00")) @@ -183,6 +196,25 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: om_today = _sf(maxtemps[0]) if maxtemps else None forecast_daily = [{"date": d, "max_temp": t} for d, t in zip(dates, maxtemps)] + if om_today is None: + nws_high = _sf(raw.get("nws", {}).get("today_high")) + mb_high = _sf(raw.get("meteoblue", {}).get("today_high")) + mgm_high = _sf(mgm.get("today_high")) if mgm else None + fallback_high = ( + nws_high + if nws_high is not None + else mb_high + if mb_high is not None + else mgm_high + if mgm_high is not None + else max_so_far + if max_so_far is not None + else cur_temp + ) + if fallback_high is not None: + om_today = float(fallback_high) + if not forecast_daily: + forecast_daily = [{"date": local_date_str, "max_temp": om_today}] sunrise = ( sunrises[0].split("T")[1][:5] if sunrises and "T" in str(sunrises[0]) @@ -271,6 +303,29 @@ def _analyze(city: str, force_refresh: bool = False) -> Dict[str, Any]: h_wdir = hourly.get("wind_direction_10m", []) h_precip_prob = hourly.get("precipitation_probability", []) h_cloud_cover = hourly.get("cloud_cover", []) + if (not h_times or not h_temps) and metar: + metar_today_obs = metar.get("today_obs", []) or [] + parsed_obs = [] + for item in metar_today_obs: + try: + t_str, t_val = item + if t_str is None or t_val is None: + continue + hh, mm = str(t_str).split(":") + parsed_obs.append((int(hh), int(mm), float(t_val))) + except Exception: + continue + if parsed_obs: + parsed_obs.sort(key=lambda x: (x[0], x[1])) + h_times = [f"{local_date_str}T{hh:02d}:{mm:02d}" for hh, mm, _ in parsed_obs] + h_temps = [v for _, _, v in parsed_obs] + h_rad = [0 for _ in parsed_obs] + h_dew = [None for _ in parsed_obs] + h_pressure = [None for _ in parsed_obs] + h_wspd = [None for _ in parsed_obs] + h_wdir = [None for _ in parsed_obs] + h_precip_prob = [None for _ in parsed_obs] + h_cloud_cover = [None for _ in parsed_obs] peak_hours = [] if h_times and h_temps and om_today is not None: