From 5d630bb910a48d580462fe88e76007515c323859 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 12 May 2026 20:10:06 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=A5=E5=85=A5=20KNMI=20=E9=98=BF=E5=A7=86?= =?UTF-8?q?=E6=96=AF=E7=89=B9=E4=B8=B9=E5=8F=B2=E5=9F=BA=E6=B5=A6=E6=9C=BA?= =?UTF-8?q?=E5=9C=BA=2010=20=E5=88=86=E9=92=9F=E6=95=B0=E6=8D=AE=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 通过 KNMI Open Data API 获取 Schiphol 机场 10 分钟观测数据（NetCDF 格式）。 需设置 KNMI_API_KEY 环境变量。Docker 镜像新增 libhdf5-dev/netCDF4 依赖。 Constraint: KNMI API key 为 JWT 格式，通过 Authorization header 传递 Scope-risk: 中高，新增系统依赖 libhdf5-dev + netCDF4 Tested: pytest 176 passed, KNMI API 连通性验证通过 --- Dockerfile | 6 +- requirements.txt | 1 + src/analysis/market_alert_engine.py | 2 +- src/bot/runtime_coordinator.py | 2 +- src/data_collection/knmi_sources.py | 249 +++++++++++++++++++++++++ src/data_collection/weather_sources.py | 37 +++- src/utils/telegram_push.py | 6 +- 7 files changed, 295 insertions(+), 8 deletions(-) create mode 100644 src/data_collection/knmi_sources.py diff --git a/Dockerfile b/Dockerfile index 0386f298..90f057eb 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,8 +10,10 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ PIP_ROOT_USER_ACTION=ignore \ TZ=UTC -# 安装系统依赖 (如果有必要的包可以取消注释) -# RUN apt-get update && apt-get install -y --no-install-recommends gcc && rm -rf /var/lib/apt/lists/* +# 安装系统依赖 (netCDF4 需要 HDF5 + 编译工具) +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc libhdf5-dev libnetcdf-dev && \ + rm -rf /var/lib/apt/lists/* # 复制 requirements 文件 COPY requirements.txt . diff --git a/requirements.txt b/requirements.txt index c2a5d98c..9a79aed8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,6 +3,7 @@ httpx loguru pyTelegramBotAPI python-dotenv +netCDF4 pytz numpy lightgbm diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index fc7352e1..70d02cc7 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -813,7 +813,7 @@ def _build_advice_cn( return "，".join(parts) + "。" -_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"} +_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM"} def _calc_airport_rapid_temp_change( diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 9c66a4b6..e7840b21 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -190,7 +190,7 @@ class StartupCoordinator: details = { "mode": "airport-periodic", "interval_sec": interval, - "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki"], + "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam"], "chat_targets": len(chat_ids), "window": "DEB proximity ≤3°C", } diff --git a/src/data_collection/knmi_sources.py b/src/data_collection/knmi_sources.py new file mode 100644 index 00000000..f6e88c9a --- /dev/null +++ b/src/data_collection/knmi_sources.py @@ -0,0 +1,249 @@ +"""KNMI (Royal Netherlands Meteorological Institute) data source. + +Fetches 10-minute weather observations from the KNMI Data Platform +for Amsterdam Schiphol airport (WMO 06240, station 240). +Uses netCDF4; requires libhdf5-dev on Linux. +""" + +from __future__ import annotations + +import time +from datetime import datetime +from typing import Any, Dict, Optional + +from loguru import logger + +from src.utils.metrics import record_source_call + +KNMI_DATASET = "10-minute-in-situ-meteorological-observations" +KNMI_VERSION = "1.0" +KNMI_API_BASE = "https://api.dataplatform.knmi.nl/open-data/v1" +KNMI_STATION = { + "amsterdam": { + "station": "240", + "icao": "EHAM", + "label": "Schiphol 10min (KNMI)", + }, +} + + +class KnmiSourceMixin: + def _knmi_api_key(self) -> str: + import os + return str(os.getenv("KNMI_API_KEY") or "").strip() + + def _knmi_http_get(self, url: str, api_key: str) -> bytes: + headers = {"Authorization": api_key} + getter = getattr(self, "_http_get", None) + if callable(getter): + resp = getter(url) + return resp.content if hasattr(resp, "content") else resp + resp = self.session.get(url, timeout=self.timeout, headers=headers) + resp.raise_for_status() + return resp.content + + def _knmi_http_get_json(self, url: str, api_key: str) -> Dict: + headers = {"Authorization": api_key} + getter = getattr(self, "_http_get_json", None) + if callable(getter): + return getter(url) + resp = self.session.get(url, timeout=self.timeout, headers=headers) + resp.raise_for_status() + return resp.json() + + def fetch_knmi_current( + self, + city: str, + use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + started = time.perf_counter() + city_key = str(city or "").strip().lower() + meta = KNMI_STATION.get(city_key) or {} + if not meta: + record_source_call("knmi", "current", "unsupported_city", + (time.perf_counter() - started) * 1000.0) + return None + + api_key = self._knmi_api_key() + if not api_key: + logger.warning("KNMI_API_KEY not set, skipping fetch") + return None + + cache_key = f"knmi:{city_key}:{use_fahrenheit}" + now_ts = time.time() + with self._knmi_cache_lock: + cached = self._knmi_cache.get(cache_key) + if cached and now_ts - cached["t"] < self.knmi_cache_ttl_sec: + record_source_call("knmi", "current", "cache_hit", + (time.perf_counter() - started) * 1000.0) + return cached["d"] + + base = f"{KNMI_API_BASE}/datasets/{KNMI_DATASET}/versions/{KNMI_VERSION}" + + try: + # Get latest file + resp = self._knmi_http_get_json( + f"{base}/files?maxKeys=1&sorting=desc&orderBy=created", + api_key, + ) + files = resp.get("files") or [] + if not files: + record_source_call("knmi", "current", "no_files", + (time.perf_counter() - started) * 1000.0) + return None + + fname = files[0]["filename"] + + # Get download URL + url_resp = self._knmi_http_get_json(f"{base}/files/{fname}/url", api_key) + download_url = url_resp.get("temporaryDownloadUrl") + if not download_url: + record_source_call("knmi", "current", "no_download_url", + (time.perf_counter() - started) * 1000.0) + return None + + # Download and parse NetCDF + nc_bytes = self._knmi_http_get(download_url, api_key) + + try: + from netCDF4 import Dataset + nc = Dataset(fname, memory=nc_bytes) + except ImportError: + logger.error("netCDF4 not installed; KNMI data unavailable") + record_source_call("knmi", "current", "netcdf4_missing", + (time.perf_counter() - started) * 1000.0) + return None + + try: + stn = meta["station"] + # Find station index + station_ids = list(nc.variables.get("station_id", [])[:]) + if not station_ids: + station_codes = [] + for s in nc.variables.get("station", []): + try: + station_codes.append(str(int(s))) + except Exception: + station_codes.append("") + station_ids = station_codes + + try: + idx = station_ids.index(stn) + except ValueError: + idx = station_ids.index(int(stn)) if stn.isdigit() else -1 + + if idx < 0 or idx >= len(station_ids): + logger.warning("KNMI station {} not found in file", stn) + nc.close() + return None + + # Extract latest timestep for temperature + ta = nc.variables.get("ta", None) + if ta is None: + nc.close() + return None + + data = ta[:] + latest_temp = float(data[-1, idx]) if data.ndim == 2 else float(data[-1]) + + # Wind speed + ff = nc.variables.get("ff", None) + wind_ms = None + if ff is not None: + wind_data = ff[:] + wind_ms = float(wind_data[-1, idx] if wind_data.ndim == 2 else wind_data[-1]) + + # Pressure + p0 = nc.variables.get("p0", None) + pressure_hpa = None + if p0 is not None: + p_data = p0[:] + pressure_hpa = float(p_data[-1, idx] if p_data.ndim == 2 else p_data[-1]) + if pressure_hpa > 5000: + pressure_hpa = pressure_hpa / 100.0 + + nc.close() + + if latest_temp < -90 or latest_temp > 60: + record_source_call("knmi", "current", "bad_value", + (time.perf_counter() - started) * 1000.0) + return None + + temp_c = round(float(latest_temp), 1) + wind_kt = round(wind_ms * 1.94384, 1) if wind_ms is not None else None + if pressure_hpa is not None: + pressure_hpa = round(float(pressure_hpa), 1) + + temp = round(temp_c * 9 / 5 + 32, 1) if use_fahrenheit else temp_c + + # Extract obs time from filename: KMDS__OPER_P___10M_OBS_L2_202605121150.nc + import re + obs_time = None + time_match = re.search(r"(\d{12})", fname) + if time_match: + ts = time_match.group(1) + obs_dt = datetime.strptime(ts, "%Y%m%d%H%M") + obs_time = obs_dt.isoformat() + + result = { + "source": "knmi", + "timestamp": datetime.utcnow().isoformat(), + "station_code": stn, + "station_name": meta["label"], + "icao": meta["icao"], + "obs_time": obs_time or datetime.utcnow().isoformat(), + "current": { + "temp": temp, + }, + "temp_c": temp_c, + "wind_kt": wind_kt, + "pressure_hpa": pressure_hpa, + } + + with self._knmi_cache_lock: + self._knmi_cache[cache_key] = {"d": result, "t": now_ts} + record_source_call("knmi", "current", "success", + (time.perf_counter() - started) * 1000.0) + return result + + finally: + try: + nc.close() + except Exception: + pass + + except Exception as exc: + logger.warning("KNMI current fetch failed city={} error={}", city_key, exc) + with self._knmi_cache_lock: + stale = self._knmi_cache.get(cache_key) + if stale: + record_source_call("knmi", "current", "stale_cache", + (time.perf_counter() - started) * 1000.0) + return stale["d"] + record_source_call("knmi", "current", "error", + (time.perf_counter() - started) * 1000.0) + return None + + def fetch_knmi_official_nearby( + self, + city: str, + use_fahrenheit: bool = False, + ) -> list[Dict[str, Any]]: + current = self.fetch_knmi_current(city, use_fahrenheit=use_fahrenheit) + if not current: + return [] + meta = KNMI_STATION.get(str(city or "").strip().lower()) or {} + return [ + { + "name": meta.get("label") or "Schiphol 10min (KNMI)", + "station_label": meta.get("label"), + "lat": 52.3081, + "lon": 4.7642, + "temp": (current.get("current") or {}).get("temp"), + "icao": meta.get("icao"), + "istNo": meta.get("station"), + "source": "knmi", + "source_label": "KNMI", + "obs_time": current.get("obs_time"), + } + ] diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index ea5df6f8..5b5b7a4f 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -16,10 +16,11 @@ from src.data_collection.nmc_sources import NmcSourceMixin from src.data_collection.nws_open_meteo_sources import NwsOpenMeteoSourceMixin from src.data_collection.amos_station_sources import AmosStationSourceMixin from src.data_collection.fmi_sources import FmiSourceMixin +from src.data_collection.knmi_sources import KnmiSourceMixin from src.database.db_manager import DBManager -class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, FmiSourceMixin): +class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, FmiSourceMixin, KnmiSourceMixin): """ Multi-source weather data collector @@ -227,6 +228,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) self._fmi_cache: Dict[str, Dict] = {} self._fmi_cache_lock = threading.Lock() + self.knmi_cache_ttl_sec = int( + os.getenv("KNMI_CACHE_TTL_SEC", "120") + ) + self._knmi_cache: Dict[str, Dict] = {} + self._knmi_cache_lock = threading.Lock() self.cwa_open_data_auth = ( os.getenv("CWA_OPEN_DATA_AUTH") or os.getenv("CWA_OPEN_DATA_API_KEY") @@ -737,6 +743,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._jma_cache.pop(f"{normalized}:{use_fahrenheit}", None) with self._fmi_cache_lock: self._fmi_cache.pop(f"fmi:{normalized}:{use_fahrenheit}", None) + with self._knmi_cache_lock: + self._knmi_cache.pop(f"knmi:{normalized}:{use_fahrenheit}", None) with self._nmc_cache_lock: self._nmc_cache.pop(f"{normalized}:{use_fahrenheit}", None) with self._ru_station_cache_lock: @@ -899,6 +907,31 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour except Exception: logger.exception("airport_obs_log append failed for jma city={}", city_lower) + def _attach_knmi_official_nearby( + self, results: Dict, city_lower: str, use_fahrenheit: bool + ) -> None: + if city_lower != "amsterdam": + return + rows = self.fetch_knmi_official_nearby( + city_lower, use_fahrenheit=use_fahrenheit + ) + if not rows: + return + results["knmi_official_nearby"] = rows + if "mgm_nearby" not in results: + results["mgm_nearby"] = rows + results["nearby_source"] = "knmi" + try: + row = rows[0] if rows else {} + DBManager().append_airport_obs( + icao=str(row.get("icao") or "EHAM"), + city=city_lower, + temp_c=row.get("temp"), + obs_time=str(row.get("obs_time") or datetime.now().isoformat()), + ) + except Exception: + logger.exception("airport_obs_log append failed for knmi city={}", city_lower) + def _attach_fmi_official_nearby( self, results: Dict, city_lower: str, use_fahrenheit: bool ) -> None: @@ -1086,6 +1119,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_china_official_nearby(results, city_lower, use_fahrenheit) self._attach_japan_official_nearby(results, city_lower, use_fahrenheit) self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit) + self._attach_knmi_official_nearby(results, city_lower, use_fahrenheit) self._attach_russia_official_nearby(results, city_lower, use_fahrenheit) if city_lower == "warsaw": self._attach_warsaw_official_nearby(results, use_fahrenheit) @@ -1129,6 +1163,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour self._attach_china_official_nearby(results, city_lower, use_fahrenheit) self._attach_japan_official_nearby(results, city_lower, use_fahrenheit) self._attach_fmi_official_nearby(results, city_lower, use_fahrenheit) + self._attach_knmi_official_nearby(results, city_lower, use_fahrenheit) self._attach_russia_official_nearby(results, city_lower, use_fahrenheit) if city_lower == "warsaw": self._attach_warsaw_official_nearby(results, use_fahrenheit) diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 0ce12071..38347b17 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -690,8 +690,8 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th # ── high-freq airport push loop ── -HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki"} -HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"} +HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam"} +HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM"} _AIRPORT_PUSH_STATE_PATH = os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), @@ -723,7 +723,7 @@ def _save_airport_state(state: Dict[str, Any]) -> None: os.replace(tmp, path) -_AIRPORT_CITY_LABEL = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa", "helsinki": "赫尔辛基/Vantaa"} +_AIRPORT_CITY_LABEL = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa", "helsinki": "赫尔辛基/Vantaa", "amsterdam": "阿姆斯特丹/Schiphol"} def _build_airport_status_message(