diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index e148f1a2..fc7352e1 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"} +_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"} def _calc_airport_rapid_temp_change( diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 16e175c8..9c66a4b6 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"], + "cities": ["seoul", "busan", "tokyo", "ankara", "helsinki"], "chat_targets": len(chat_ids), "window": "DEB proximity ≤3°C", } diff --git a/src/data_collection/fmi_sources.py b/src/data_collection/fmi_sources.py new file mode 100644 index 00000000..a64318d8 --- /dev/null +++ b/src/data_collection/fmi_sources.py @@ -0,0 +1,179 @@ +"""FMI (Finnish Meteorological Institute) open data source. + +Fetches 10-minute airport weather observations from opendata.fmi.fi +for Helsinki-Vantaa airport (EFHK, WMO 2974, FMISID 100968). +""" + +from __future__ import annotations + +import re +import time +from datetime import datetime +from typing import Any, Dict, Optional + +from loguru import logger + +from src.utils.metrics import record_source_call + +FMI_BASE_URL = "https://opendata.fmi.fi/wfs" +FMI_STATION = { + "helsinki": { + "fmisid": "100968", + "wmo": "2974", + "icao": "EFHK", + "label": "Helsinki-Vantaa 10min (FMI)", + "query_place": "helsinki-vantaa_airport", + }, +} + + +class FmiSourceMixin: + def _fmi_http_get_text(self, url: str) -> str: + getter = getattr(self, "_http_get", None) + if callable(getter): + response = getter(url) + else: + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.text + + def fetch_fmi_current( + self, + city: str, + use_fahrenheit: bool = False, + ) -> Optional[Dict[str, Any]]: + started = time.perf_counter() + city_key = str(city or "").strip().lower() + meta = FMI_STATION.get(city_key) or {} + if not meta: + record_source_call("fmi", "current", "unsupported_city", + (time.perf_counter() - started) * 1000.0) + return None + + cache_key = f"fmi:{city_key}:{use_fahrenheit}" + now_ts = time.time() + with self._fmi_cache_lock: + cached = self._fmi_cache.get(cache_key) + if cached and now_ts - cached["t"] < self.fmi_cache_ttl_sec: + record_source_call("fmi", "current", "cache_hit", + (time.perf_counter() - started) * 1000.0) + return cached["d"] + + try: + from datetime import timezone as tz + + place = meta["query_place"] + end_time = datetime.now(tz.utc) + start_time = end_time.replace(minute=end_time.minute // 10 * 10, second=0, microsecond=0) + end_str = end_time.strftime("%Y-%m-%dT%H:%M:%SZ") + start_str = (start_time - __import__("datetime").timedelta(minutes=20)).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + + url = ( + f"{FMI_BASE_URL}?service=WFS&version=2.0.0&request=getFeature" + f"&storedquery_id=fmi::observations::weather::timevaluepair" + f"&place={place}" + f"¶meters=t2m,ws_10min,p_sea" + f"&starttime={start_str}&endtime={end_str}" + ) + + xml_text = self._fmi_http_get_text(url) + + # Parse per-parameter observation blocks + blocks = re.split(r".*?(.*?)\s*(.*?)", + block, + re.DOTALL, + ) + if tvps: + latest_time, latest_val = tvps[-1] + obs_time = latest_time + try: + latest_values[param] = float(latest_val) + except (ValueError, TypeError): + pass + + temp_c = latest_values.get("t2m") + wind_ms = latest_values.get("ws_10min") + pressure_hpa = latest_values.get("p_sea") + + if temp_c is None: + record_source_call("fmi", "current", "no_temperature", + (time.perf_counter() - started) * 1000.0) + return None + + # Convert wind from m/s to kt + wind_kt = round(wind_ms * 1.94384, 1) if wind_ms is not None else None + + temp = round(temp_c * 9 / 5 + 32, 1) if use_fahrenheit else round(temp_c, 1) + + result = { + "source": "fmi", + "timestamp": datetime.utcnow().isoformat(), + "station_code": meta["fmisid"], + "station_name": meta["label"], + "wmo": meta["wmo"], + "icao": meta["icao"], + "obs_time": obs_time, + "current": { + "temp": temp, + }, + "temp_c": temp_c, + "wind_kt": wind_kt, + "pressure_hpa": pressure_hpa, + } + + with self._fmi_cache_lock: + self._fmi_cache[cache_key] = {"d": result, "t": now_ts} + record_source_call("fmi", "current", "success", + (time.perf_counter() - started) * 1000.0) + return result + + except Exception as exc: + logger.warning("FMI current fetch failed city={} error={}", city_key, exc) + with self._fmi_cache_lock: + stale = self._fmi_cache.get(cache_key) + if stale: + record_source_call("fmi", "current", "stale_cache", + (time.perf_counter() - started) * 1000.0) + return stale["d"] + record_source_call("fmi", "current", "error", + (time.perf_counter() - started) * 1000.0) + return None + + def fetch_fmi_official_nearby( + self, + city: str, + use_fahrenheit: bool = False, + ) -> list[Dict[str, Any]]: + current = self.fetch_fmi_current(city, use_fahrenheit=use_fahrenheit) + if not current: + return [] + meta = FMI_STATION.get(str(city or "").strip().lower()) or {} + return [ + { + "name": meta.get("label") or "Helsinki-Vantaa 10min (FMI)", + "station_label": meta.get("label"), + "lat": 60.32937, + "lon": 24.97274, + "temp": (current.get("current") or {}).get("temp"), + "icao": meta.get("icao"), + "istNo": meta.get("fmisid"), + "source": "fmi", + "source_label": "FMI", + "obs_time": current.get("obs_time"), + } + ] diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index c5acb04d..ea5df6f8 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -15,10 +15,11 @@ from src.data_collection.russia_station_sources import RussiaStationSourceMixin 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.database.db_manager import DBManager -class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin): +class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin, FmiSourceMixin): """ Multi-source weather data collector @@ -221,6 +222,11 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour ) self._settlement_cache: Dict[str, Dict] = {} self._settlement_cache_lock = threading.Lock() + self.fmi_cache_ttl_sec = int( + os.getenv("FMI_CACHE_TTL_SEC", "120") + ) + self._fmi_cache: Dict[str, Dict] = {} + self._fmi_cache_lock = threading.Lock() self.cwa_open_data_auth = ( os.getenv("CWA_OPEN_DATA_AUTH") or os.getenv("CWA_OPEN_DATA_API_KEY") @@ -729,6 +735,8 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour normalized = str(city or "").strip().lower() with self._jma_cache_lock: 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._nmc_cache_lock: self._nmc_cache.pop(f"{normalized}:{use_fahrenheit}", None) with self._ru_station_cache_lock: @@ -891,6 +899,32 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour except Exception: logger.exception("airport_obs_log append failed for jma city={}", city_lower) + def _attach_fmi_official_nearby( + self, results: Dict, city_lower: str, use_fahrenheit: bool + ) -> None: + if city_lower != "helsinki": + return + rows = self.fetch_fmi_official_nearby( + city_lower, use_fahrenheit=use_fahrenheit + ) + if not rows: + return + results["fmi_official_nearby"] = rows + if "mgm_nearby" not in results: + results["mgm_nearby"] = rows + results["nearby_source"] = "fmi" + try: + row = rows[0] if rows else {} + icao = str(row.get("icao") or "EFHK") + DBManager().append_airport_obs( + icao=icao, + 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 fmi city={}", city_lower) + def _attach_korean_amos_data( self, results: Dict, city_lower: str, use_fahrenheit: bool ) -> None: @@ -1051,6 +1085,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if include_nearby: 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_russia_official_nearby(results, city_lower, use_fahrenheit) if city_lower == "warsaw": self._attach_warsaw_official_nearby(results, use_fahrenheit) @@ -1093,6 +1128,7 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if include_nearby: 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_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 58efb5f0..0ce12071 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"} -HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128"} +HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki"} +HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK"} _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"} +_AIRPORT_CITY_LABEL = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa", "helsinki": "赫尔辛基/Vantaa"} def _build_airport_status_message(