From 05a5f005feb3b57b70709cc43595d12f80971012 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Fri, 6 Mar 2026 10:51:06 +0800 Subject: [PATCH] feat: add a rule-based market alert engine for weather analysis with Telegram integration and tests. --- src/analysis/market_alert_engine.py | 120 ++++++++++++++++++++++++++++ src/utils/telegram_push.py | 8 ++ tests/test_market_alert_engine.py | 26 ++++++ 3 files changed, 154 insertions(+) diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index a555648a..ba4243f7 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -381,6 +381,84 @@ def _pick_leading_station(city: str, nearby: List[Dict[str, Any]]) -> Optional[D return max(nearby, key=_temp) +def _pick_ankara_center_station(nearby: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: + if not nearby: + return None + + def _temp(row: Dict[str, Any]) -> float: + return _sf(row.get("temp")) or -999.0 + + priority_rows = [] + for row in nearby: + name = str(row.get("name") or "").lower() + sid = str(row.get("istNo") or "").strip() + if sid == "17130" or "center" in name or "b枚lge" in name or "etimesgut" in name: + priority_rows.append(row) + if priority_rows: + return max(priority_rows, key=_temp) + return None + + +def _calc_ankara_center_deb_alert( + city_weather: Dict[str, Any], + temp_symbol: str, +) -> Dict[str, Any]: + city = (city_weather.get("name") or "").lower() + if city != "ankara": + return { + "type": "ankara_center_deb_hit", + "triggered": False, + "reason": "city is not ankara", + } + + deb_prediction = _sf((city_weather.get("deb") or {}).get("prediction")) + if deb_prediction is None: + return { + "type": "ankara_center_deb_hit", + "triggered": False, + "reason": "deb prediction unavailable", + } + + center_station = _pick_ankara_center_station(city_weather.get("mgm_nearby") or []) + if not center_station: + return { + "type": "ankara_center_deb_hit", + "triggered": False, + "reason": "ankara center station unavailable", + } + + center_temp = _sf(center_station.get("temp")) + if center_temp is None: + return { + "type": "ankara_center_deb_hit", + "triggered": False, + "reason": "ankara center temperature unavailable", + } + + airport_temp = _sf((city_weather.get("current") or {}).get("temp")) + epsilon = _to_unit_delta(0.05, temp_symbol) + triggered = center_temp + epsilon >= deb_prediction + + return { + "type": "ankara_center_deb_hit", + "triggered": triggered, + "force_push": triggered, + "center_station": { + "name": center_station.get("name"), + "istNo": center_station.get("istNo"), + "temp": round(center_temp, 2), + }, + "deb_prediction": round(deb_prediction, 2), + "airport_temp": round(airport_temp, 2) if airport_temp is not None else None, + "margin_vs_deb": round(center_temp - deb_prediction, 2), + "center_lead_vs_airport": ( + round(center_temp - airport_temp, 2) + if airport_temp is not None + else None + ), + } + + def _calc_advection_alert(city_weather: Dict[str, Any], temp_symbol: str) -> Dict[str, Any]: city = (city_weather.get("name") or "").lower() current = city_weather.get("current") or {} @@ -470,6 +548,7 @@ def _calc_advection_alert(city_weather: Dict[str, Any], temp_symbol: str) -> Dic def _join_trigger_types_cn(rules: Dict[str, Dict[str, Any]]) -> str: mapping = [ + ("ankara_center_deb_hit", "Center达到DEB"), ("momentum_spike", "动量突变"), ("forecast_breakthrough", "预测突破"), ("kill_zone", "临界触发"), @@ -484,11 +563,21 @@ def _build_advice_cn( temp_symbol: str, ) -> str: parts: List[str] = [] + center_deb = rules.get("ankara_center_deb_hit", {}) advection = rules.get("advection", {}) momentum = rules.get("momentum_spike", {}) breakthrough = rules.get("forecast_breakthrough", {}) kill_zone = rules.get("kill_zone", {}) + if center_deb.get("triggered"): + deb_prediction = _sf(center_deb.get("deb_prediction")) + center_temp = _sf(((center_deb.get("center_station") or {}).get("temp"))) + if deb_prediction is not None and center_temp is not None: + parts.append( + f"Ankara Center {center_temp:.1f}{temp_symbol} 已触及 DEB {deb_prediction:.1f}{temp_symbol}" + ) + else: + parts.append("Ankara Center 已触及 DEB 预测值") if advection.get("triggered"): parts.append("风向转南，暖平流增强") if momentum.get("triggered"): @@ -520,6 +609,7 @@ def _build_telegram_messages( temp_symbol = city_weather.get("temp_symbol", "°C") city_name = city_weather.get("display_name") or city_weather.get("name", "").title() current_temp = _sf((city_weather.get("current") or {}).get("temp")) + center_deb = rules.get("ankara_center_deb_hit", {}) momentum = rules.get("momentum_spike", {}) kill_zone = rules.get("kill_zone", {}) advection = rules.get("advection", {}) @@ -535,6 +625,7 @@ def _build_telegram_messages( distance = _sf(kill_zone.get("distance")) market_label = str(kill_zone.get("market_label") or "").strip() market_prices = kill_zone.get("market_prices") or {} + center_station = center_deb.get("center_station") or {} dyn = f"实测 {current_temp:.1f}{temp_symbol}" if delta_temp is not None and delta_min is not None: @@ -555,6 +646,22 @@ def _build_telegram_messages( if lead_delta is not None: lead_line = f"联动：{st_name} 已领先 {lead_delta:+.1f}{temp_symbol}" + center_deb_line = "" + if center_deb.get("triggered"): + center_name = center_station.get("name") or "Ankara Center" + center_temp = _sf(center_station.get("temp")) + deb_prediction = _sf(center_deb.get("deb_prediction")) + airport_temp = _sf(center_deb.get("airport_temp")) + lead_gap = _sf(center_deb.get("center_lead_vs_airport")) + if center_temp is not None and deb_prediction is not None: + center_deb_line = ( + f"Center信号：{center_name} {center_temp:.1f}{temp_symbol} 已达到 DEB {deb_prediction:.1f}{temp_symbol}" + ) + if airport_temp is not None: + center_deb_line += f" | 机场 {airport_temp:.1f}{temp_symbol}" + if lead_gap is not None: + center_deb_line += f" | 领先 {lead_gap:+.1f}{temp_symbol}" + price_line = "" if any( market_prices.get(key) is not None @@ -581,6 +688,8 @@ def _build_telegram_messages( ] if strike_line: lines_zh.append(strike_line) + if center_deb_line: + lines_zh.append(center_deb_line) if price_line: lines_zh.append(price_line) if lead_line: @@ -607,6 +716,13 @@ def _build_telegram_messages( ] if strike is not None and distance is not None: lines_en.append(f"Distance to {strike:.1f}{temp_symbol} strike: {distance:.1f}{temp_symbol}") + if center_deb_line: + center_temp = _sf(center_station.get("temp")) + deb_prediction = _sf(center_deb.get("deb_prediction")) + if center_temp is not None and deb_prediction is not None: + lines_en.append( + f"Center signal: {center_temp:.1f}{temp_symbol} has reached DEB {deb_prediction:.1f}{temp_symbol}" + ) if price_line: price_label_en = f"Quotes ({market_label}): " if market_label else "Quotes: " lines_en.append( @@ -634,6 +750,7 @@ def build_trading_alerts( now = datetime.now(timezone.utc).isoformat() rules: Dict[str, Dict[str, Any]] = { + "ankara_center_deb_hit": _calc_ankara_center_deb_alert(city_weather, temp_symbol), "momentum_spike": _calc_momentum_alert(city_weather, temp_symbol), "forecast_breakthrough": _calc_forecast_breakthrough_alert(city_weather, temp_symbol), "kill_zone": _calc_kill_zone_alert(city_weather, market_snapshot, temp_symbol), @@ -648,7 +765,10 @@ def build_trading_alerts( for key, value in rules.items() if value.get("triggered") ] + force_push = any(alert.get("force_push") for alert in triggered) severity = "high" if len(triggered) >= 2 else ("medium" if len(triggered) == 1 else "none") + if force_push and severity == "none": + severity = "medium" telegram = _build_telegram_messages( city_weather=city_weather, diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index fd2d43ca..fbf12f09 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -100,6 +100,10 @@ def _cleanup_state(state: Dict[str, Any], now_ts: int, keep_sec: int = 7 * 86400 def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_count: int) -> bool: + triggered_alerts = alert_payload.get("triggered_alerts") or [] + if any(alert.get("force_push") for alert in triggered_alerts): + return True + trigger_count = int(alert_payload.get("trigger_count") or 0) if trigger_count < min_trigger_count: return False @@ -109,6 +113,7 @@ def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_c def _alert_signature(alert_payload: Dict[str, Any]) -> str: rules = alert_payload.get("rules") or {} + center_deb = rules.get("ankara_center_deb_hit") or {} momentum = rules.get("momentum_spike") or {} breakthrough = rules.get("forecast_breakthrough") or {} kill_zone = rules.get("kill_zone") or {} @@ -123,6 +128,9 @@ def _alert_signature(alert_payload: Dict[str, Any]) -> str: for alert in (alert_payload.get("triggered_alerts") or []) if alert.get("type") ), + "center_temp": round(float(((center_deb.get("center_station") or {}).get("temp")) or 0.0), 1), + "center_deb_prediction": round(float(center_deb.get("deb_prediction") or 0.0), 1), + "center_airport_gap": round(float(center_deb.get("center_lead_vs_airport") or 0.0), 1), "momentum_direction": momentum.get("direction"), "momentum_slope_30m": round(float(momentum.get("slope_30m") or 0.0), 1), "breakthrough_margin": round(float(breakthrough.get("margin") or 0.0), 1), diff --git a/tests/test_market_alert_engine.py b/tests/test_market_alert_engine.py index 8221838f..0645b4ce 100644 --- a/tests/test_market_alert_engine.py +++ b/tests/test_market_alert_engine.py @@ -25,6 +25,9 @@ def _sample_weather_payload(): "GFS": 10.4, "ECMWF": 10.6, }, + "deb": { + "prediction": 11.8, + }, "metar_recent_obs": [ {"time": "10:30", "wdir": 180}, {"time": "10:00", "wdir": 60}, @@ -100,3 +103,26 @@ def test_forecast_breakthrough_not_triggered_when_current_not_above_margin(): market_snapshot=_sample_market_snapshot(), ) assert out["rules"]["forecast_breakthrough"]["triggered"] is False + + +def test_ankara_center_hits_deb_triggers_force_push(): + city_weather = _sample_weather_payload() + city_weather["current"]["temp"] = 10.7 + city_weather["deb"]["prediction"] = 11.2 + city_weather["trend"]["recent"] = [ + {"time": "10:30", "temp": 10.7}, + {"time": "10:00", "temp": 10.7}, + {"time": "09:30", "temp": 10.6}, + ] + city_weather["multi_model"] = {"MGM": 11.2, "GFS": 11.2, "ECMWF": 11.2} + + out = build_trading_alerts( + city_weather=city_weather, + market_snapshot={"city": "ankara", "target_date": "2026-03-07", "markets": []}, + ) + + center_rule = out["rules"]["ankara_center_deb_hit"] + assert center_rule["triggered"] is True + assert center_rule["force_push"] is True + assert out["severity"] in ("medium", "high") + assert "Center信号" in out["telegram"]["zh"]