feat: Implement PolyWeather web map API by reusing existing analysis logic and centralize weather data collection.

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")
                         # 计算数据年龄

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(600, self._disk_cache_max_age_sec)
-            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
             loaded = 0
-            for key, entry in saved.get("forecast", {}).items():
+            with self._open_meteo_cache_lock:
-                if now - float(entry.get("t", 0)) < max_age:
+                for key, entry in saved.get("forecast", {}).items():
-                    self._open_meteo_cache[key] = entry
+                    if now - float(entry.get("t", 0)) < max_age:
-                    loaded += 1
+                        old = self._open_meteo_cache.get(key)
-            for key, entry in saved.get("ensemble", {}).items():
+                        if old is None or float(entry.get("t", 0)) >= float(old.get("t", 0)):
-                if now - float(entry.get("t", 0)) < max_age:
+                            self._open_meteo_cache[key] = entry
-                    self._ensemble_cache[key] = entry
+                            loaded += 1
-                    loaded += 1
+            with self._ensemble_cache_lock:
-            for key, entry in saved.get("multi_model", {}).items():
+                for key, entry in saved.get("ensemble", {}).items():
-                if now - float(entry.get("t", 0)) < max_age:
+                    if now - float(entry.get("t", 0)) < max_age:
-                    self._multi_model_cache[key] = entry
+                        old = self._ensemble_cache.get(key)
-                    loaded += 1
+                        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:

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 预设的静态偏移 (兜底当地时间)
+    # 优先从 API 获取偏移；若缺失则尝试 NWS 动态偏移；最后回退静态配置
-    utc_offset = om.get("utc_offset", info.get("tz", 0))
+    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: