diff --git a/docs/AIRPORT_MARKET_MONITOR_ZH.md b/docs/AIRPORT_MARKET_MONITOR_ZH.md index 15dee466..5b5a64fa 100644 --- a/docs/AIRPORT_MARKET_MONITOR_ZH.md +++ b/docs/AIRPORT_MARKET_MONITOR_ZH.md @@ -4,11 +4,12 @@ ### 现有数据 -| 城市 | 机场 | ICAO | 数据源 | 刷新频率 | 缓存 TTL | 数据字段 | -|------|------|------|--------|---------|----------|---------| +| 城市 | 机场 | ICAO/站点 | 数据源 | 刷新频率 | 缓存 TTL | 数据字段 | +|------|------|-----------|--------|---------|----------|---------| | 首尔 | 仁川国际 | RKSI | AMOS (`global.amo.go.kr`) | 1 分钟 | 60s | 温度/露点/风速/气压/能见度/RVR/跑道对温度/METAR/TAF | | 釜山 | 金海国际 | RKPK | AMOS (`global.amo.go.kr`) | 1 分钟 | 60s | 同上 | -| 东京 | 羽田 | RJTT | JMA AMeDAS (`jma.go.jp`) | 10 分钟 | 120s | 10 分钟温度观测（当前仅提取 temp） | +| 东京 | 羽田 | RJTT | JMA AMeDAS (`jma.go.jp`) | 10 分钟 | 120s | 10 分钟温度观测 | +| 安卡拉 | Esenboğa | 17128 | MGM (`servis.mgm.gov.tr`) | ~10 分钟 | 无独立缓存 | 温度/风速/气压/湿度/降水 | ### 现有 Telegram 推送系统 @@ -20,9 +21,7 @@ ### 问题 -高频机场数据已就绪，但现有推送系统 30 分钟一轮对所有城市一视同仁。首尔/釜山 1 分钟级 AMOS 和东京 10 分钟级 JMA 接近交易高峰期时，温度变化可能比 30 分钟窗口更快，需要更灵敏的监控。 - -同时，DEB 每日最高温预测是结算的核心参照，将机场实时温度与 DEB 预测并排对比，可以快速发现实际温度偏离预测的程度。 +四座机场城市的实时数据已就绪，但现有推送系统 30 分钟一轮对所有城市一视同仁。1-10 分钟级高频数据在接近交易高峰期时，温度变化可能比 30 分钟窗口更快，需要更灵敏的监控。 --- @@ -30,15 +29,15 @@ ### 核心思路 -**在现有 30 分钟主循环之上叠加高频通道**，对三大机场城市用更短间隔推送"实时温度 + DEB 预测最高温"快报。不做市场分析、不输出 AI 建议，只报温度数据本身。 +在现有 30 分钟主循环之上叠加高频通道，对四座机场城市用 10 分钟间隔独立检测温度急变。温度波动达到阈值时推送告警，包含当前温度 + DEB 预测最高温。不做市场分析、不输出 AI 建议、不约定时快照。 ### 1. 高频机场城市快速通道 -在现有 30 分钟主循环之外，为 `{seoul, busan, tokyo}` 单独跑一个 10 分钟间隔的子循环，仅检查温度动量突变。 +在现有 30 分钟主循环之外，为 `{seoul, busan, tokyo, ankara}` 单独跑一个 10 分钟间隔的子循环，每个城市独立检测温度急变。 **配置（写死在代码中）**: ```python -HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo"} +HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara"} HIGH_FREQ_PUSH_INTERVAL_SEC = 600 # 10 分钟 HIGH_FREQ_MOMENTUM_THRESHOLD_C = 0.5 # 比默认 0.8°C 更灵敏 HIGH_FREQ_COOLDOWN_SEC = 7200 # 同一城市冷却 2 小时 @@ -46,15 +45,13 @@ HIGH_FREQ_COOLDOWN_SEC = 7200 # 同一城市冷却 2 小时 **逻辑**: - 主循环 30 分钟照常跑全部城市（不变） -- 每 10 分钟额外跑一轮高频城市子集 -- 高频轮次仅检查温度动量突变一条规则 -- 高频告警独立冷却期 2 小时 -- **最高温已锁定则跳过**：当日最高温已出现且之后持续下降，不再推送 +- 每 10 分钟对四座机场城市各检查一次温度急变 +- 高频轮次仅检查 `airport_rapid_temp_change` 一条规则 +- 各城市独立冷却，触发后 2 小时内同一城市不再重复推送 +- **最高温已锁定则跳过**：当日最高已过且持续下降，不再推送 ### 2. 机场观测积累与趋势检测 -**问题**: 当前 AMOS/JMA 每次只返回最新一条观测，无法计算温度斜率做动量检测。 - **新增数据库表**: `airport_obs_log` ```sql CREATE TABLE IF NOT EXISTS airport_obs_log ( @@ -71,52 +68,29 @@ CREATE INDEX IF NOT EXISTS idx_airport_obs_log_icao_time ON airport_obs_log(icao, created_at DESC); ``` -**写入**: 在 AMOS/JMA 成功获取数据后自动调用 `append_airport_obs()` 写入。自动清理 2 小时前的旧数据。 +**写入**: 在 AMOS/JMA/MGM 成功获取数据后自动调用 `append_airport_obs()` 写入。自动清理 2 小时前的旧数据。 **读取**: `get_airport_obs_recent(icao, minutes=30)` 返回最近 N 分钟观测列表，用于计算温度变化斜率。 ### 3. 温度突变即时告警 -基于积累的观测日志，新增告警规则 `airport_rapid_temp_change`。 +基于积累的观测日志，新增告警规则 `airport_rapid_temp_change`。每条告警 per-city 独立推送。 | 参数 | 值 | 说明 | |------|-----|------| | 滑动窗口 | 20 分钟 | 取最近 20 分钟内的观测 | | 最少样本 | 3 条 | 确保有足够数据点 | -| 触发阈值 | > 0.5°C/10min | 高频通道专用，比默认 0.8°C 更灵敏 | +| 触发阈值 | > 0.5°C/10min | 比默认 0.8°C/30min 更灵敏 | +| 冷却期 | 2 小时 | 同城市两次推送最小间隔 | | 锁定跳过 | 最高温已锁定 | 当日最高已过且持续下降，不推送 | **告警消息示例**: ``` -🚨 首尔/仁川 温度急变 +🚨 首尔/仁川 温度急变 🔥 -跑道中位数 14.9°C → 16.5°C (+1.6°C / 15min) -风 220° 14kt 暖平流增强 -DEB 预测今日最高 18.2°C | 目前距预测差 1.7°C -``` - -### 4. 机场实况快照摘要（含 DEB 预测） - -每隔 30 分钟，向市场监控频道推送三座机场的"当前温度 + DEB 预测最高温"摘要。 - -**推送时段**: 仅各城市当地 08:00-20:00（避免半夜噪音） - -**锁定跳过**: 如该城市当日最高温已锁定，跳过该城市不推送（三城中仍有未锁定的则正常推送其余） - -**消息格式**: -``` -🛫 机场实况 14:30 CST - -🇰🇷 首尔/仁川 RKSI -跑道中位数 14.9°C · DEB 预测最高 18.2°C -风 220° 14kt | QNH 1015.2 hPa | 能见度 ≥10km - -🇰🇷 釜山/金海 RKPK -跑道中位数 13.8°C · DEB 预测最高 16.5°C -风 180° 8kt - -🇯🇵 东京/羽田 RJTT -当前 12.4°C (14:20 JST) · DEB 预测最高 15.1°C +跑道温度 14.9°C → 16.5°C (+1.6°C / 15min) +风 14kt +DEB 预测今日最高 18.2°C | 差距 +1.7°C ``` --- @@ -126,26 +100,26 @@ DEB 预测今日最高 18.2°C | 目前距预测差 1.7°C | 优先级 | 文件 | 改动 | |--------|------|------| | 1 | `src/database/db_manager.py` | 新增 `airport_obs_log` 表、`append_airport_obs()`、`get_airport_obs_recent()` | -| 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` | 高频快速通道子循环、机场快照推送（含 DEB 预测）、温度急变告警集成 | +| 2 | `src/data_collection/weather_sources.py` | AMOS/JMA/MGM 成功后调用 `append_airport_obs()` 写日志 | +| 3 | `src/analysis/market_alert_engine.py` | 新增 `airport_rapid_temp_change` 规则 | +| 4 | `src/utils/telegram_push.py` | 10 分钟高频子循环、温度急变告警推送、最高温锁定跳过 | +| 5 | `src/bot/runtime_coordinator.py` | 注册机场高频推送循环 | --- ## 实施顺序 1. **Phase 1 — DB 层**: `airport_obs_log` 表 + 读写方法 -2. **Phase 2 — 采集层**: AMOS/JMA 成功后自动写日志，部署观察 1-2 天确认数据积累正常 +2. **Phase 2 — 采集层**: AMOS/JMA/MGM 成功后自动写日志，部署观察 1-2 天确认数据积累正常 3. **Phase 3 — 告警引擎**: `airport_rapid_temp_change` 规则 + 单元测试 -4. **Phase 4 — 推送层**: 高频快速通道 + 机场快照 + DEB 预测，直接推送市场监控频道 +4. **Phase 4 — 推送层**: 高频快速通道，直接推送市场监控频道 5. **Phase 5 — 调参**: 观察 3-7 天调整阈值 --- ## 风险与注意事项 -- **AMOS/JMA 站点可用性**: `global.amo.go.kr` 和 `jma.go.jp` 可能偶发性不可用，需容错处理 +- **AMOS/JMA/MGM 站点可用性**: 各数据源可能偶发性不可用，需容错处理 - **告警频率控制**: 高频循环可能产生过多告警，需要严格的冷却期和去重机制 -- **数据库体积**: `airport_obs_log` 每 1-10 分钟写入 3 条记录，2 小时约 36-360 条，自动清理后体积可控 -- **东京 JMA 数据完整性**: 当前只提取 temp，风速/气压需要额外解析（JMA JSON 中有 `wind` 和 `pressure` 数组但未使用） +- **数据库体积**: `airport_obs_log` 每 1-10 分钟写入 4 条记录，2 小时约 48-480 条，自动清理后体积可控 +- **安卡拉 MGM 刷新频率**: `servis.mgm.gov.tr` 实际更新间隔待确认，暂按 ~10 分钟预估 diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index 1dfe5cce..5de84f63 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"} +_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128"} def _calc_airport_rapid_temp_change( diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index 6cf4dd4e..c5acb04d 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -805,6 +805,19 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour if not mgm_data: return results["mgm"] = mgm_data + # Log airport obs for high-freq monitoring (Ankara 17128) + try: + mgm_current = mgm_data.get("current") or {} + DBManager().append_airport_obs( + icao=str(istno), + city=city_lower, + temp_c=mgm_current.get("temp"), + wind_kt=mgm_current.get("wind_speed_kt"), + pressure_hpa=mgm_current.get("pressure"), + obs_time=mgm_data.get("obs_time") or datetime.now().isoformat(), + ) + except Exception: + logger.exception("airport_obs_log append failed for mgm city={}", city_lower) if include_nearby: results["nearby_source"] = "mgm" nearby = self.fetch_mgm_nearby_stations(province, root_ist_no=istno) diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 1bcaf0bb..8f0227f1 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -690,12 +690,11 @@ 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"} -HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT"} +HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara"} +HIGH_FREQ_AIRPORT_ICAO = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128"} 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__)))), @@ -758,7 +757,7 @@ def _build_airport_rapid_change_message( deb_pred: Optional[float], ) -> str: city_display = city.title() - airport_label = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田"}.get(city, city_display) + airport_label = {"seoul": "首尔/仁川", "busan": "釜山/金海", "tokyo": "东京/羽田", "ankara": "安卡拉/Esenboğa"}.get(city, city_display) emoji = "🔥" if rule.get("direction") == "up" else "❄️" first = rule.get("first_temp", 0) last = rule.get("last_temp", 0) @@ -781,42 +780,6 @@ def _build_airport_rapid_change_message( 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") - runway_pairs = s.get("runway_pairs") or [] - runway_temps = s.get("runway_temps") or [] - lines.append("") - header = f"{flag.get(city, '')} {name.get(city, city)}" - - # Show individual runway pair temps for Seoul/Busan - if runway_pairs and runway_temps and len(runway_pairs) == len(runway_temps): - runway_lines = [] - for (r1, r2), (t, d) in zip(runway_pairs, runway_temps): - dew_str = f" / 露点 {d:.1f}°C" if d is not None else "" - runway_lines.append(f"{r1}/{r2}: {t:.1f}°C{dew_str}") - line = header + "\n" + "\n".join(runway_lines) - elif temp is not None: - line = header + f"\n当前 {temp:.1f}°C" - else: - line = header - - if deb is not None: - line += f"\nDEB 预测最高 {deb:.1f}°C" - 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], @@ -827,9 +790,6 @@ def _run_high_freq_airport_cycle( 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, "") @@ -844,7 +804,6 @@ def _run_high_freq_airport_cycle( except Exception: latest_obs = {} - # Fetch city_weather once and extract DEB city_weather: Dict[str, Any] = {} deb_pred: Optional[float] = None try: @@ -856,51 +815,13 @@ def _run_high_freq_airport_cycle( except Exception: pass - # Extract airport-level data from city_weather for snapshot - amos = city_weather.get("amos") or {} - runway_data = (amos.get("runway_obs") or {}) if amos else {} - runway_pairs = runway_data.get("runway_pairs") or [] - runway_temps = runway_data.get("temperatures") or [] - # Tokyo JMA data lives in mgm_nearby (set by _attach_japan_official_nearby) - mgm_nearby = city_weather.get("mgm_nearby") or [] - nearby_row = mgm_nearby[0] if mgm_nearby else {} - current_temp = ( - amos.get("temp_c") - or nearby_row.get("temp") - or latest_obs.get("temp_c") - or (city_weather.get("current") or {}).get("temp") - ) - wind_kt = amos.get("wind_kt") or latest_obs.get("wind_kt") - pressure_hpa = amos.get("pressure_hpa") or latest_obs.get("pressure_hpa") - # Check high-locked if _check_high_locked(city): - if snapshot_due: - snapshots.append({ - "city": city, - "temp_c": current_temp, - "wind_kt": wind_kt, - "pressure_hpa": pressure_hpa, - "deb_prediction": deb_pred, - "runway_pairs": runway_pairs, - "runway_temps": runway_temps, - }) 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": current_temp, - "wind_kt": wind_kt, - "pressure_hpa": pressure_hpa, - "deb_prediction": deb_pred, - "runway_pairs": runway_pairs, - "runway_temps": runway_temps, - }) - if not rule.get("triggered"): continue @@ -926,24 +847,6 @@ def _run_high_freq_airport_cycle( 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: - kst_now = datetime.utcnow() + timedelta(hours=9) - local_time = kst_now.strftime("%H:%M KST") - 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