From 64f8ff21ec9952a30a66c895e1b533cc02576f9e Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Tue, 12 May 2026 17:04:17 +0800 Subject: [PATCH] =?UTF-8?q?@=20=E6=96=B0=E5=A2=9E=E6=9C=BA=E5=9C=BA?= =?UTF-8?q?=E9=AB=98=E9=A2=91=E6=8E=A8=E9=80=81=EF=BC=9A10=E5=88=86?= =?UTF-8?q?=E9=92=9F=E7=BA=A7=E6=B8=A9=E5=BA=A6=E7=9B=91=E6=8E=A7=20+=20DE?= =?UTF-8?q?B=E9=A2=84=E6=B5=8B=E5=BF=AB=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为首尔/釜山/东京三大机场城市实现高频温度监控通道： - 新增 airport_obs_log 表积累观测数据，支持趋势检测 - AMOS/JMA 成功后自动写入观测日志 - 新增 airport_rapid_temp_change 告警规则（20min窗口、0.5°C/10min阈值） - 10分钟间隔高频子循环 + 30分钟机场快照（含 DEB 预测最高温） - 最高温锁定后自动跳过，快照仅当地 08:00-20:00 发送 Tested: ruff check 通过 @ --- docs/AIRPORT_MARKET_MONITOR_ZH.md | 55 ++--- src/analysis/market_alert_engine.py | 79 +++++++ src/bot/runtime_coordinator.py | 25 +++ src/data_collection/weather_sources.py | 23 ++ src/database/db_manager.py | 55 +++++ src/utils/telegram_push.py | 284 +++++++++++++++++++++++++ 6 files changed, 496 insertions(+), 25 deletions(-) diff --git a/docs/AIRPORT_MARKET_MONITOR_ZH.md b/docs/AIRPORT_MARKET_MONITOR_ZH.md index 0e92f649..15dee466 100644 --- a/docs/AIRPORT_MARKET_MONITOR_ZH.md +++ b/docs/AIRPORT_MARKET_MONITOR_ZH.md @@ -1,4 +1,4 @@ -# 机场高频数据接入 Telegram 市场监控频道方案 +# 机场高频数据接入市场监控频道方案 ## 背景 @@ -15,12 +15,14 @@ - **循环**: `start_trade_alert_push_loop`，默认每 30 分钟跑一轮 - **覆盖城市**: `TELEGRAM_ALERT_CITIES`（默认全部 51 城） - **4 条规则**: Ankara Center DEB 命中、动量突变(30min 斜率 > 0.8°C)、预报突破、暖平流 -- **门禁**: 严重度/触发数/市场可交易性/冷却期 多层过滤 -- **消息**: 中英双语，包含触发类型、实况温度、市场概率分布、AI 建议 +- **门禁**: 严重度/触发数/冷却期 多层过滤 +- **消息**: 中英双语，包含触发类型、实况温度 ### 问题 -高频机场数据已就绪，但现有推送系统没有针对性利用。首尔/釜山 1 分钟级 AMOS 和东京 10 分钟级 JMA 可以做更快、更灵敏的市场信号检测。 +高频机场数据已就绪，但现有推送系统 30 分钟一轮对所有城市一视同仁。首尔/釜山 1 分钟级 AMOS 和东京 10 分钟级 JMA 接近交易高峰期时，温度变化可能比 30 分钟窗口更快，需要更灵敏的监控。 + +同时，DEB 每日最高温预测是结算的核心参照，将机场实时温度与 DEB 预测并排对比，可以快速发现实际温度偏离预测的程度。 --- @@ -28,24 +30,26 @@ ### 核心思路 -**增强现有系统，不新建**。对高频机场城市使用更低的推送间隔和更灵敏的规则参数。 +**在现有 30 分钟主循环之上叠加高频通道**，对三大机场城市用更短间隔推送"实时温度 + DEB 预测最高温"快报。不做市场分析、不输出 AI 建议，只报温度数据本身。 ### 1. 高频机场城市快速通道 -在现有 30 分钟主循环之外，为 `{seoul, busan, tokyo}` 单独跑一个 10 分钟间隔的子循环，只检查动量突变和预报突破两条最关键的规则，使用更低的触发阈值。 +在现有 30 分钟主循环之外，为 `{seoul, busan, tokyo}` 单独跑一个 10 分钟间隔的子循环，仅检查温度动量突变。 **配置（写死在代码中）**: ```python HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo"} HIGH_FREQ_PUSH_INTERVAL_SEC = 600 # 10 分钟 HIGH_FREQ_MOMENTUM_THRESHOLD_C = 0.5 # 比默认 0.8°C 更灵敏 +HIGH_FREQ_COOLDOWN_SEC = 7200 # 同一城市冷却 2 小时 ``` **逻辑**: - 主循环 30 分钟照常跑全部城市（不变） - 每 10 分钟额外跑一轮高频城市子集 -- 高频轮次跳过 Ankara DEB 和 Advection 规则 -- 高频告警有独立冷却期（如 2 小时，比默认 6 小时更短） +- 高频轮次仅检查温度动量突变一条规则 +- 高频告警独立冷却期 2 小时 +- **最高温已锁定则跳过**：当日最高温已出现且之后持续下降，不再推送 ### 2. 机场观测积累与趋势检测 @@ -79,39 +83,40 @@ CREATE INDEX IF NOT EXISTS idx_airport_obs_log_icao_time |------|-----|------| | 滑动窗口 | 20 分钟 | 取最近 20 分钟内的观测 | | 最少样本 | 3 条 | 确保有足够数据点 | -| 触发阈值 | > 0.3°C/10min | 约 1.8°C/小时 | -| 额外条件 | 温度变化方向与市场桶一致 | 避免无关告警 | +| 触发阈值 | > 0.5°C/10min | 高频通道专用，比默认 0.8°C 更灵敏 | +| 锁定跳过 | 最高温已锁定 | 当日最高已过且持续下降，不推送 | **告警消息示例**: ``` 🚨 首尔/仁川 温度急变 -跑道中位数 15.2→16.8°C (+1.6°C / 15min) -风 180°→210° 暖平流增强 -市场桶 17°C 概率 62% → 当前边缘 +8.2% +跑道中位数 14.9°C → 16.5°C (+1.6°C / 15min) +风 220° 14kt 暖平流增强 +DEB 预测今日最高 18.2°C | 目前距预测差 1.7°C ``` -### 4. 机场实况快照摘要 +### 4. 机场实况快照摘要（含 DEB 预测） -每隔 30 分钟，向市场监控频道推送三座机场的当前实况摘要。 +每隔 30 分钟，向市场监控频道推送三座机场的"当前温度 + DEB 预测最高温"摘要。 **推送时段**: 仅各城市当地 08:00-20:00（避免半夜噪音） +**锁定跳过**: 如该城市当日最高温已锁定，跳过该城市不推送（三城中仍有未锁定的则正常推送其余） + **消息格式**: ``` -🛫 机场实况快照 14:30 CST +🛫 机场实况 14:30 CST 🇰🇷 首尔/仁川 RKSI -跑道 15L/33R: 14.6°C | 15R/33L: 15.2°C -风 220° 14kt | QNH 1015.2 hPa | 能见度 ≥10km -METAR 14:00Z | 跑道中位数 14.9°C +跑道中位数 14.9°C · DEB 预测最高 18.2°C +风 220° 14kt | QNH 1015.2 hPa | 能见度 ≥10km 🇰🇷 釜山/金海 RKPK -跑道 18L/36R: 13.8°C | 风 180° 8kt -METAR 14:00Z +跑道中位数 13.8°C · DEB 预测最高 16.5°C +风 180° 8kt 🇯🇵 东京/羽田 RJTT -JMA AMeDAS 10-min: 12.4°C (14:20 JST) +当前 12.4°C (14:20 JST) · DEB 预测最高 15.1°C ``` --- @@ -124,7 +129,7 @@ JMA AMeDAS 10-min: 12.4°C (14:20 JST) | 2 | `src/data_collection/amos_station_sources.py` | 成功获取后调用 `append_airport_obs()` 写日志 | | 3 | `src/data_collection/jma_amedas_sources.py` | 成功获取后调用 `append_airport_obs()` 写日志，可选扩展提取更多字段（风速/气压） | | 4 | `src/analysis/market_alert_engine.py` | 新增 `airport_rapid_temp_change` 规则 | -| 5 | `src/utils/telegram_push.py` | 高频快速通道子循环、机场快照推送、温度急变告警集成 | +| 5 | `src/utils/telegram_push.py` | 高频快速通道子循环、机场快照推送（含 DEB 预测）、温度急变告警集成 | --- @@ -133,8 +138,8 @@ JMA AMeDAS 10-min: 12.4°C (14:20 JST) 1. **Phase 1 — DB 层**: `airport_obs_log` 表 + 读写方法 2. **Phase 2 — 采集层**: AMOS/JMA 成功后自动写日志，部署观察 1-2 天确认数据积累正常 3. **Phase 3 — 告警引擎**: `airport_rapid_temp_change` 规则 + 单元测试 -4. **Phase 4 — 推送层**: 高频快速通道 + 机场快照，先在测试频道验证消息格式 -5. **Phase 5 — 上线**: 切换到正式市场监控频道，观察 3-7 天调整阈值 +4. **Phase 4 — 推送层**: 高频快速通道 + 机场快照 + DEB 预测，直接推送市场监控频道 +5. **Phase 5 — 调参**: 观察 3-7 天调整阈值 --- diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index 3cb1ef9a..1dfe5cce 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -10,6 +10,7 @@ from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from src.analysis.settlement_rounding import apply_city_settlement +from src.database.db_manager import DBManager def _sf(v: Any) -> Optional[float]: @@ -812,6 +813,83 @@ def _build_advice_cn( return "，".join(parts) + "。" +_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT"} + + +def _calc_airport_rapid_temp_change( + city_weather: Dict[str, Any], temp_symbol: str +) -> Dict[str, Any]: + city = (city_weather.get("name") or "").lower() + if city not in _AIRPORT_ICAO_MAP: + return { + "type": "airport_rapid_temp_change", + "triggered": False, + "reason": "not_airport_city", + } + + icao = _AIRPORT_ICAO_MAP[city] + try: + db = DBManager() + obs = db.get_airport_obs_recent(icao, minutes=20) + except Exception: + return { + "type": "airport_rapid_temp_change", + "triggered": False, + "reason": "db_read_error", + } + + if len(obs) < 3: + return { + "type": "airport_rapid_temp_change", + "triggered": False, + "reason": f"insufficient_obs ({len(obs)}<3)", + } + + first = obs[0] + last = obs[-1] + first_temp = first.get("temp_c") + last_temp = last.get("temp_c") + if first_temp is None or last_temp is None: + return { + "type": "airport_rapid_temp_change", + "triggered": False, + "reason": "missing_temp", + } + + try: + t1 = datetime.fromisoformat(str(first.get("created_at") or "")) + t2 = datetime.fromisoformat(str(last.get("created_at") or "")) + delta_min = (t2 - t1).total_seconds() / 60.0 + except Exception: + delta_min = len(obs) * 1.5 + + if delta_min <= 0: + return { + "type": "airport_rapid_temp_change", + "triggered": False, + "reason": "zero_time_delta", + } + + delta_temp = last_temp - first_temp + slope_per_10min = delta_temp / delta_min * 10.0 + threshold = _to_unit_delta(0.5, temp_symbol) + triggered = abs(slope_per_10min) > threshold + + return { + "type": "airport_rapid_temp_change", + "triggered": triggered, + "direction": "up" if slope_per_10min > 0 else "down", + "first_temp": round(first_temp, 2), + "last_temp": round(last_temp, 2), + "delta_temp": round(delta_temp, 2), + "delta_min": round(delta_min, 1), + "slope_per_10min": round(slope_per_10min, 2), + "threshold": round(threshold, 2), + "sample_count": len(obs), + "icao": icao, + } + + def _build_telegram_messages( city_weather: Dict[str, Any], rules: Dict[str, Dict[str, Any]], @@ -1179,6 +1257,7 @@ def build_trading_alerts( "momentum_spike": _calc_momentum_alert(city_weather, temp_symbol), "forecast_breakthrough": _calc_forecast_breakthrough_alert(city_weather, temp_symbol), "advection": _calc_advection_alert(city_weather, temp_symbol), + "airport_rapid_temp_change": _calc_airport_rapid_temp_change(city_weather, temp_symbol), } triggered = [ diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 940a84dc..3fa3ac7c 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -84,6 +84,7 @@ class StartupCoordinator: def start_all(self) -> RuntimeStatus: loops = [ self._start_trade_alert_loop(), + self._start_airport_high_freq_loop(), self._start_dashboard_prewarm_loop(), self._start_polygon_wallet_loop(), self._start_polymarket_wallet_activity_loop(), @@ -182,6 +183,30 @@ class StartupCoordinator: ), ) + def _start_airport_high_freq_loop(self) -> LoopStatus: + enabled = _env_bool("TELEGRAM_AIRPORT_PUSH_ENABLED", True) + chat_ids = get_telegram_chat_ids_from_env() + interval = max(60, _env_int("TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC", 600)) + details = { + "mode": "airport-high-freq", + "interval_sec": interval, + "cities": ["seoul", "busan", "tokyo"], + "chat_targets": len(chat_ids), + "cooldown_sec": _env_int("TELEGRAM_AIRPORT_COOLDOWN_SEC", 7200), + } + validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS" + return self._start_with_validation( + key="airport_high_freq_push", + label="机场高频推送", + configured_enabled=enabled, + details=details, + validation_error=validation_error, + starter=lambda: import_module("src.utils.telegram_push").start_high_freq_airport_push_loop( + self.bot, + self.config, + ), + ) + def _start_dashboard_prewarm_loop(self) -> LoopStatus: enabled = _env_bool("POLYWEATHER_DASHBOARD_PREWARM_ENABLED", False) interval = max(30, _env_int("POLYWEATHER_PREWARM_INTERVAL_SEC", 300)) diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 4c884aca..6cf4dd4e 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -15,6 +15,7 @@ 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.database.db_manager import DBManager class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSourceMixin, MgmSourceMixin, JmaAmedasSourceMixin, RussiaStationSourceMixin, NmcSourceMixin, NwsOpenMeteoSourceMixin, AmosStationSourceMixin): @@ -865,6 +866,17 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if "mgm_nearby" not in results: results["mgm_nearby"] = official_rows results["nearby_source"] = "jma" + try: + row = official_rows[0] if official_rows else {} + icao = str(row.get("icao") or row.get("istNo") or "RJTT") + 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 jma city={}", city_lower) def _attach_korean_amos_data( self, results: Dict, city_lower: str, use_fahrenheit: bool @@ -882,6 +894,17 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour city_lower, amos_data.get("temp_c"), amos_data.get("temp_source"), len(amos_data.get("runway_obs", {}).get("runway_pairs", []) or [])) results["amos"] = amos_data + try: + DBManager().append_airport_obs( + icao=amos_data.get("icao") or "", + city=city_lower, + temp_c=amos_data.get("temp_c"), + wind_kt=amos_data.get("wind_kt"), + pressure_hpa=amos_data.get("pressure_hpa"), + obs_time=amos_data.get("observation_time") or datetime.now().isoformat(), + ) + except Exception: + logger.exception("airport_obs_log append failed for amos city={}", city_lower) else: logger.warning("AMOS: no data returned for city={}", city_lower) except Exception as exc: diff --git a/src/database/db_manager.py b/src/database/db_manager.py index 9ab5b0dd..e68f2444 100644 --- a/src/database/db_manager.py +++ b/src/database/db_manager.py @@ -280,6 +280,21 @@ class DBManager: conn.execute( "CREATE INDEX IF NOT EXISTS idx_supabase_bindings_telegram_id ON supabase_bindings(telegram_id)" ) + conn.execute(""" + CREATE TABLE IF NOT EXISTS airport_obs_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + icao TEXT NOT NULL, + city TEXT NOT NULL, + temp_c REAL, + wind_kt REAL, + pressure_hpa REAL, + obs_time TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_airport_obs_log_icao_time ON airport_obs_log(icao, created_at DESC)" + ) self._ensure_column(conn, "users", "daily_points", "INTEGER DEFAULT 0") self._ensure_column(conn, "users", "daily_points_date", "TEXT") self._ensure_column(conn, "users", "weekly_points", "INTEGER DEFAULT 0") @@ -1825,3 +1840,43 @@ class DBManager: ) conn.commit() return True + + def append_airport_obs( + self, + *, + icao: str, + city: str, + temp_c: Optional[float] = None, + wind_kt: Optional[float] = None, + pressure_hpa: Optional[float] = None, + obs_time: str, + ) -> None: + with self._get_connection() as conn: + conn.execute( + """ + INSERT INTO airport_obs_log (icao, city, temp_c, wind_kt, pressure_hpa, obs_time) + VALUES (?, ?, ?, ?, ?, ?) + """, + (str(icao).strip().upper(), str(city).strip().lower(), + temp_c, wind_kt, pressure_hpa, str(obs_time)), + ) + conn.execute( + "DELETE FROM airport_obs_log WHERE created_at < datetime('now', '-2 hours')" + ) + conn.commit() + + def get_airport_obs_recent( + self, icao: str, minutes: int = 30 + ) -> list[dict[str, Any]]: + with self._get_connection() as conn: + conn.row_factory = sqlite3.Row + rows = conn.execute( + """ + SELECT icao, city, temp_c, wind_kt, pressure_hpa, obs_time, created_at + FROM airport_obs_log + WHERE icao = ? AND created_at >= datetime('now', ? || ' minutes') + ORDER BY created_at ASC + """, + (str(icao).strip().upper(), str(-int(minutes))), + ).fetchall() + return [dict(r) for r in rows] diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 8c4429a4..3d30e063 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -686,3 +686,287 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th ) thread.start() return thread + + +# ── high-freq airport push loop ── + +HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo"} +HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT"} +HIGH_FREQ_PUSH_INTERVAL_SEC = 600 +HIGH_FREQ_MOMENTUM_THRESHOLD_C = 0.5 +HIGH_FREQ_COOLDOWN_SEC = 7200 +AIRPORT_SNAPSHOT_INTERVAL_SEC = 1800 + +_AIRPORT_PUSH_STATE_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), + "data", "airport_push_state.json", +) + + +def _load_airport_state() -> Dict[str, Any]: + try: + from src.database.runtime_state import STATE_STORAGE_SQLITE, get_state_storage_mode + if get_state_storage_mode() == STATE_STORAGE_SQLITE: + return _telegram_state_repo.load_state() + except Exception: + pass + path = _AIRPORT_PUSH_STATE_PATH + if not os.path.exists(path): + return {"last_by_city": {}} + try: + with open(path, "r", encoding="utf-8") as fh: + data = json.load(fh) + if isinstance(data, dict): + data.setdefault("last_by_city", {}) + return data + except Exception: + pass + return {"last_by_city": {}} + + +def _save_airport_state(state: Dict[str, Any]) -> None: + try: + from src.database.runtime_state import STATE_STORAGE_SQLITE, get_state_storage_mode + if get_state_storage_mode() == STATE_STORAGE_SQLITE: + _telegram_state_repo.save_state(state) + return + except Exception: + pass + path = _AIRPORT_PUSH_STATE_PATH + os.makedirs(os.path.dirname(path), exist_ok=True) + tmp = f"{path}.tmp" + with open(tmp, "w", encoding="utf-8") as fh: + json.dump(state, fh, ensure_ascii=False, indent=2) + os.replace(tmp, path) + + +def _check_high_locked(city: str) -> bool: + icao = HIGH_FREQ_AIRPORT_ICAO.get(city) + if not icao: + return False + try: + from src.database.db_manager import DBManager + db = DBManager() + obs = db.get_airport_obs_recent(icao, minutes=120) + except Exception: + return False + if len(obs) < 6: + return False + temps = [r.get("temp_c") for r in obs if r.get("temp_c") is not None] + if len(temps) < 6: + return False + peak_idx = temps.index(max(temps)) + if peak_idx >= len(temps) - 2: + return False + post_peak = temps[peak_idx:] + for i in range(1, len(post_peak)): + if post_peak[i] > post_peak[i - 1] + 0.05: + return False + return True + + +def _build_airport_rapid_change_message( + city: str, + rule: Dict[str, Any], + deb_pred: Optional[float], +) -> str: + city_display = city.title() + airport_label = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田"}.get(city, city_display) + emoji = "🔥" if rule.get("direction") == "up" else "❄️" + first = rule.get("first_temp", 0) + last = rule.get("last_temp", 0) + delta = rule.get("delta_temp", 0) + delta_min = rule.get("delta_min", 0) + sign = "+" if delta >= 0 else "" + + lines = [ + f"🚨 {airport_label} 温度急变 {emoji}", + "", + f"跑道温度 {first:.1f}°C → {last:.1f}°C ({sign}{delta:.1f}°C / {delta_min:.0f}min)", + ] + wind_kt = rule.get("wind_kt") + if wind_kt is not None: + lines.append(f"风 {wind_kt:.0f}kt") + if deb_pred is not None: + gap = last - deb_pred + gap_sign = "+" if gap >= 0 else "" + lines.append(f"DEB 预测今日最高 {deb_pred:.1f}°C | 差距 {gap_sign}{gap:.1f}°C") + return "\n".join(lines) + + +def _build_airport_snapshot_message(snapshots: list[dict[str, Any]], local_time: str = "") -> str: + flag = {"seoul": "🇰🇷", "busan": "🇰🇷", "tokyo": "🇯🇵"} + name = {"seoul": "首尔/仁川 RKSI", "busan": "釜山/金海 RKPK", "tokyo": "东京/羽田 RJTT"} + + lines = [f"🛫 机场实况 {local_time}"] + for s in snapshots: + city = s.get("city", "") + temp = s.get("temp_c") + deb = s.get("deb_prediction") + lines.append("") + header = f"{flag.get(city, '')} {name.get(city, city)}" + parts = [] + if temp is not None: + parts.append(f"当前 {temp:.1f}°C") + if deb is not None: + parts.append(f"DEB 预测最高 {deb:.1f}°C") + line = header + if parts: + line += "\n" + " · ".join(parts) + if s.get("wind_kt") is not None: + line += f" | 风 {s['wind_kt']:.0f}kt" + if s.get("pressure_hpa") is not None: + line += f" | QNH {s['pressure_hpa']:.1f} hPa" + lines.append(line) + return "\n".join(lines) + + +def _run_high_freq_airport_cycle( + bot: Any, + config: Dict[str, Any], + chat_ids: List[str], + state: Dict[str, Any], +) -> bool: + state_dirty = False + now_ts = int(time.time()) + last_by_city = state.setdefault("last_by_city", {}) + + snapshots: list[dict[str, Any]] = [] + snapshot_due = now_ts - int(state.get("last_snapshot_ts") or 0) >= AIRPORT_SNAPSHOT_INTERVAL_SEC + + for city in sorted(HIGH_FREQ_AIRPORT_CITIES): + try: + icao = HIGH_FREQ_AIRPORT_ICAO.get(city, "") + last_city = last_by_city.get(city) or {} + last_city_ts = int(last_city.get("ts") or 0) + + try: + from src.database.db_manager import DBManager + db = DBManager() + recent_obs = db.get_airport_obs_recent(icao, minutes=20) + latest_obs = recent_obs[-1] if recent_obs else {} + except Exception: + latest_obs = {} + + # Fetch city_weather once and extract DEB + city_weather: Dict[str, Any] = {} + deb_pred: Optional[float] = None + try: + from web.app import _analyze + city_weather = _analyze(city) + deb_raw = (city_weather.get("deb") or {}).get("prediction") + if deb_raw is not None: + deb_pred = float(deb_raw) + except Exception: + pass + + # Check high-locked + if _check_high_locked(city): + if snapshot_due: + snapshots.append({ + "city": city, + "temp_c": latest_obs.get("temp_c"), + "wind_kt": latest_obs.get("wind_kt"), + "pressure_hpa": latest_obs.get("pressure_hpa"), + "deb_prediction": deb_pred, + }) + continue + + from src.analysis.market_alert_engine import _calc_airport_rapid_temp_change + rule = _calc_airport_rapid_temp_change(city_weather, "°C") + + if snapshot_due: + snapshots.append({ + "city": city, + "temp_c": latest_obs.get("temp_c"), + "wind_kt": latest_obs.get("wind_kt"), + "pressure_hpa": latest_obs.get("pressure_hpa"), + "deb_prediction": deb_pred, + }) + + if not rule.get("triggered"): + continue + + if last_city_ts and now_ts - last_city_ts < HIGH_FREQ_COOLDOWN_SEC: + continue + + rule["wind_kt"] = latest_obs.get("wind_kt") + message = _build_airport_rapid_change_message(city, rule, deb_pred) + + sent = False + for chat_id in chat_ids: + try: + bot.send_message(chat_id, message) + sent = True + except Exception as exc: + logger.warning("airport push failed city={} chat_id={}: {}", city, chat_id, exc) + + if sent: + last_by_city[city] = {"ts": now_ts, "trigger": "rapid_temp_change"} + state_dirty = True + logger.info("airport rapid change pushed city={} delta_temp={:.1f}", city, rule.get("delta_temp", 0)) + + except Exception: + logger.exception("high freq airport cycle failed for city={}", city) + + if snapshot_due and snapshots: + from datetime import timedelta + kst_hour = (datetime.utcnow() + timedelta(hours=9)).hour + if 8 <= kst_hour < 20: + local_time = datetime.now().strftime("%H:%M CST") + snap_message = _build_airport_snapshot_message(snapshots, local_time) + for chat_id in chat_ids: + try: + bot.send_message(chat_id, snap_message) + except Exception as exc: + logger.warning("airport snapshot push failed chat_id={}: {}", chat_id, exc) + logger.info("airport snapshot pushed cities={}", len(snapshots)) + else: + logger.info("airport snapshot skipped: outside 08:00-20:00 window (KST hour={})", kst_hour) + state["last_snapshot_ts"] = now_ts + state_dirty = True + + return state_dirty + + +def start_high_freq_airport_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[threading.Thread]: + enabled = _env_bool("TELEGRAM_AIRPORT_PUSH_ENABLED", True) + chat_ids = get_telegram_chat_ids_from_env() + if not enabled: + logger.info("airport high-freq push loop disabled") + return None + if not chat_ids: + logger.warning("airport high-freq push loop skipped: TELEGRAM_CHAT_IDS is not set") + return None + + interval_sec = max(60, _env_int("TELEGRAM_AIRPORT_PUSH_INTERVAL_SEC", HIGH_FREQ_PUSH_INTERVAL_SEC)) + + def _runner() -> None: + state = _load_airport_state() + logger.info( + "airport high-freq push loop started cities={} interval={}s chat_targets={}", + len(HIGH_FREQ_AIRPORT_CITIES), interval_sec, len(chat_ids), + ) + while True: + cycle_started = time.time() + state = _load_airport_state() + if _run_high_freq_airport_cycle( + bot=bot, + config=config, + chat_ids=chat_ids, + state=state, + ): + _save_airport_state(state) + + elapsed = time.time() - cycle_started + sleep_sec = max(5, interval_sec - int(elapsed)) + time.sleep(sleep_sec) + + thread = threading.Thread( + target=_runner, + name="airport-high-freq-pusher", + daemon=True, + ) + thread.start() + logger.info("airport high-freq push loop thread started") + return thread