diff --git a/docker-compose.yml b/docker-compose.yml index 083c5ab6..648a6380 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -45,68 +45,3 @@ services: - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data user: "${UID:-1000}:${GID:-1000}" - polyweather_prometheus: - image: prom/prometheus:v3.4.1 - container_name: polyweather_prometheus - restart: unless-stopped - profiles: ["monitoring"] - depends_on: - - polyweather_web - command: - - "--config.file=/etc/prometheus/prometheus.yml" - - "--storage.tsdb.path=/prometheus" - - "--storage.tsdb.retention.time=15d" - - "--web.enable-lifecycle" - volumes: - - ./monitoring/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml:ro - - ./monitoring/prometheus/alerts.yml:/etc/prometheus/alerts.yml:ro - - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}/monitoring/prometheus:/prometheus - ports: - - "${POLYWEATHER_PROMETHEUS_PORT:-9090}:9090" - - polyweather_alertmanager: - image: prom/alertmanager:v0.28.1 - container_name: polyweather_alertmanager - restart: unless-stopped - profiles: ["monitoring"] - depends_on: - - polyweather_alert_relay - command: - - "--config.file=/etc/alertmanager/alertmanager.yml" - - "--storage.path=/alertmanager" - volumes: - - ./monitoring/alertmanager/alertmanager.yml:/etc/alertmanager/alertmanager.yml:ro - - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}/monitoring/alertmanager:/alertmanager - ports: - - "${POLYWEATHER_ALERTMANAGER_PORT:-9093}:9093" - - polyweather_alert_relay: - <<: *polyweather-base - container_name: polyweather_alert_relay - restart: unless-stopped - profiles: ["monitoring"] - command: python scripts/alertmanager_telegram_relay.py - volumes: - - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/var/lib/polyweather - - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}:/app/data - ports: - - "${POLYWEATHER_ALERT_RELAY_PORT:-9099}:9099" - user: "${UID:-1000}:${GID:-1000}" - - polyweather_grafana: - image: grafana/grafana-oss:12.0.2 - container_name: polyweather_grafana - restart: unless-stopped - profiles: ["monitoring"] - depends_on: - - polyweather_prometheus - environment: - GF_SECURITY_ADMIN_USER: ${POLYWEATHER_GRAFANA_ADMIN_USER:-admin} - GF_SECURITY_ADMIN_PASSWORD: ${POLYWEATHER_GRAFANA_ADMIN_PASSWORD:-polyweather} - GF_USERS_ALLOW_SIGN_UP: "false" - volumes: - - ./monitoring/grafana/provisioning:/etc/grafana/provisioning:ro - - ./monitoring/grafana/dashboards:/var/lib/grafana/dashboards:ro - - ${POLYWEATHER_RUNTIME_DATA_DIR:-/var/lib/polyweather}/monitoring/grafana:/var/lib/grafana - ports: - - "${POLYWEATHER_GRAFANA_PORT:-3001}:3000" diff --git a/monitoring/alertmanager/alertmanager.yml b/monitoring/alertmanager/alertmanager.yml deleted file mode 100644 index 274b3e10..00000000 --- a/monitoring/alertmanager/alertmanager.yml +++ /dev/null @@ -1,15 +0,0 @@ -global: - resolve_timeout: 5m - -route: - receiver: telegram-relay - group_by: ["alertname"] - group_wait: 30s - group_interval: 5m - repeat_interval: 3h - -receivers: - - name: telegram-relay - webhook_configs: - - url: http://polyweather_alert_relay:9099/alerts - send_resolved: true diff --git a/monitoring/grafana/dashboards/polyweather-overview.json b/monitoring/grafana/dashboards/polyweather-overview.json deleted file mode 100644 index 14bfe5b6..00000000 --- a/monitoring/grafana/dashboards/polyweather-overview.json +++ /dev/null @@ -1,227 +0,0 @@ -{ - "annotations": { - "list": [ - { - "builtIn": 1, - "datasource": { - "type": "grafana", - "uid": "-- Grafana --" - }, - "enable": true, - "hide": true, - "iconColor": "rgba(0, 211, 255, 1)", - "name": "Annotations & Alerts", - "type": "dashboard" - } - ] - }, - "editable": true, - "fiscalYearStartMonth": 0, - "graphTooltip": 0, - "id": null, - "links": [], - "panels": [ - { - "datasource": { - "type": "prometheus", - "uid": "prometheus" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "unit": "reqps" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 0 - }, - "id": 1, - "options": { - "legend": { - "displayMode": "table", - "placement": "bottom", - "showLegend": true - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "expr": "sum by (status) (rate(polyweather_http_requests_total[5m]))", - "legendFormat": "{{status}}", - "refId": "A" - } - ], - "title": "HTTP Requests by Status", - "type": "timeseries" - }, - { - "datasource": { - "type": "prometheus", - "uid": "prometheus" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "unit": "ms" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 0 - }, - "id": 2, - "options": { - "legend": { - "displayMode": "table", - "placement": "bottom", - "showLegend": true - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "expr": "sum(rate(polyweather_http_request_duration_ms_sum[5m])) / clamp_min(sum(rate(polyweather_http_request_duration_ms_count[5m])), 1)", - "legendFormat": "avg", - "refId": "A" - }, - { - "expr": "max(polyweather_http_request_duration_ms_max)", - "legendFormat": "max", - "refId": "B" - } - ], - "title": "HTTP Latency", - "type": "timeseries" - }, - { - "datasource": { - "type": "prometheus", - "uid": "prometheus" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "palette-classic" - }, - "unit": "reqps" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 0, - "y": 8 - }, - "id": 3, - "options": { - "legend": { - "displayMode": "table", - "placement": "bottom", - "showLegend": true - }, - "tooltip": { - "mode": "single" - } - }, - "targets": [ - { - "expr": "sum by (source, outcome) (rate(polyweather_source_requests_total[5m]))", - "legendFormat": "{{source}} / {{outcome}}", - "refId": "A" - } - ], - "title": "Source Requests by Outcome", - "type": "timeseries" - }, - { - "datasource": { - "type": "prometheus", - "uid": "prometheus" - }, - "fieldConfig": { - "defaults": { - "color": { - "mode": "thresholds" - }, - "thresholds": { - "mode": "absolute", - "steps": [ - { - "color": "green" - }, - { - "color": "orange", - "value": 10 - }, - { - "color": "red", - "value": 25 - } - ] - }, - "unit": "percentunit" - }, - "overrides": [] - }, - "gridPos": { - "h": 8, - "w": 12, - "x": 12, - "y": 8 - }, - "id": 4, - "options": { - "reduceOptions": { - "calcs": [ - "lastNotNull" - ], - "fields": "", - "values": false - } - }, - "targets": [ - { - "expr": "(sum(increase(polyweather_source_requests_total{outcome!~\"success|cache_hit\"}[15m])) / clamp_min(sum(increase(polyweather_source_requests_total[15m])), 1))", - "refId": "A" - } - ], - "title": "Source Error Rate (15m)", - "type": "stat" - } - ], - "refresh": "30s", - "schemaVersion": 41, - "style": "dark", - "tags": [ - "polyweather", - "ops" - ], - "templating": { - "list": [] - }, - "time": { - "from": "now-6h", - "to": "now" - }, - "timepicker": {}, - "timezone": "browser", - "title": "PolyWeather Overview", - "uid": "polyweather-overview", - "version": 1 -} diff --git a/monitoring/grafana/provisioning/dashboards/dashboard.yml b/monitoring/grafana/provisioning/dashboards/dashboard.yml deleted file mode 100644 index a0db257f..00000000 --- a/monitoring/grafana/provisioning/dashboards/dashboard.yml +++ /dev/null @@ -1,11 +0,0 @@ -apiVersion: 1 - -providers: - - name: PolyWeather Dashboards - orgId: 1 - folder: PolyWeather - type: file - disableDeletion: false - editable: true - options: - path: /var/lib/grafana/dashboards diff --git a/monitoring/grafana/provisioning/datasources/prometheus.yml b/monitoring/grafana/provisioning/datasources/prometheus.yml deleted file mode 100644 index f83c9fae..00000000 --- a/monitoring/grafana/provisioning/datasources/prometheus.yml +++ /dev/null @@ -1,9 +0,0 @@ -apiVersion: 1 - -datasources: - - name: PolyWeather Prometheus - type: prometheus - access: proxy - url: http://polyweather_prometheus:9090 - isDefault: true - editable: false diff --git a/monitoring/prometheus/alerts.yml b/monitoring/prometheus/alerts.yml deleted file mode 100644 index e9775a8b..00000000 --- a/monitoring/prometheus/alerts.yml +++ /dev/null @@ -1,58 +0,0 @@ -groups: - - name: polyweather-runtime - rules: - - alert: PolyWeatherWebDown - expr: up{job="polyweather-web"} == 0 - for: 2m - labels: - severity: critical - annotations: - summary: "PolyWeather web metrics endpoint is down" - description: "/metrics on polyweather_web has been unreachable for more than 2 minutes." - - - alert: PolyWeatherHttp5xxBurst - expr: sum(increase(polyweather_http_requests_total{status=~"5.."}[10m])) > 10 - for: 5m - labels: - severity: warning - annotations: - summary: "PolyWeather HTTP 5xx burst" - description: "More than 10 server errors were observed in the last 10 minutes." - - - alert: PolyWeatherHighSourceErrorRate - expr: | - ( - sum(increase(polyweather_source_requests_total{outcome!~"success|cache_hit"}[15m])) - / - clamp_min(sum(increase(polyweather_source_requests_total[15m])), 1) - ) > 0.25 - and sum(increase(polyweather_source_requests_total[15m])) > 20 - for: 10m - labels: - severity: warning - annotations: - summary: "PolyWeather source error rate is high" - description: "External weather source errors exceeded 25% over the last 15 minutes." - - - alert: PolyWeatherOpenMeteoCooldownLoop - expr: sum(increase(polyweather_source_requests_total{source="open_meteo",outcome=~"cooldown_skip|error"}[15m])) > 20 - for: 10m - labels: - severity: warning - annotations: - summary: "Open-Meteo is rate-limited or erroring" - description: "Open-Meteo has produced repeated cooldown skips or errors in the last 15 minutes." - - - alert: PolyWeatherSlowHttpAverage - expr: | - ( - sum(rate(polyweather_http_request_duration_ms_sum[10m])) - / - clamp_min(sum(rate(polyweather_http_request_duration_ms_count[10m])), 1) - ) > 2000 - for: 10m - labels: - severity: warning - annotations: - summary: "PolyWeather average HTTP latency is high" - description: "Average HTTP request duration exceeded 2s over the last 10 minutes." diff --git a/monitoring/prometheus/prometheus.yml b/monitoring/prometheus/prometheus.yml deleted file mode 100644 index b6563f44..00000000 --- a/monitoring/prometheus/prometheus.yml +++ /dev/null @@ -1,19 +0,0 @@ -global: - scrape_interval: 30s - evaluation_interval: 30s - -rule_files: - - /etc/prometheus/alerts.yml - -alerting: - alertmanagers: - - static_configs: - - targets: - - polyweather_alertmanager:9093 - -scrape_configs: - - job_name: polyweather-web - metrics_path: /metrics - static_configs: - - targets: - - polyweather_web:8000 diff --git a/scripts/alertmanager_telegram_relay.py b/scripts/alertmanager_telegram_relay.py deleted file mode 100644 index 6d6f4e14..00000000 --- a/scripts/alertmanager_telegram_relay.py +++ /dev/null @@ -1,107 +0,0 @@ -from __future__ import annotations - -import json -import os -from http.server import BaseHTTPRequestHandler, HTTPServer -from typing import Any, Dict, List - -import requests - - -def _chat_ids() -> List[str]: - raw = ( - os.getenv("POLYWEATHER_MONITORING_ALERT_CHAT_IDS") - or os.getenv("TELEGRAM_CHAT_IDS") - or os.getenv("TELEGRAM_CHAT_ID") - or "" - ) - return [item.strip() for item in raw.split(",") if item.strip()] - - -def _format_alerts(payload: Dict[str, Any]) -> str: - alerts = payload.get("alerts") or [] - if not isinstance(alerts, list) or not alerts: - return "PolyWeather monitoring received an empty alert payload." - lines = ["PolyWeather monitoring alert"] - for alert in alerts[:10]: - if not isinstance(alert, dict): - continue - status = str(alert.get("status") or "unknown").upper() - labels = alert.get("labels") or {} - annotations = alert.get("annotations") or {} - alert_name = labels.get("alertname") or "unknown_alert" - severity = labels.get("severity") or "info" - summary = annotations.get("summary") or annotations.get("description") or "" - lines.append(f"- [{status}] {alert_name} ({severity})") - if summary: - lines.append(f" {summary}") - return "\n".join(lines) - - -def _send_telegram_message(text: str) -> None: - token = str(os.getenv("TELEGRAM_BOT_TOKEN") or "").strip() - chat_ids = _chat_ids() - if not token or not chat_ids: - return - url = f"https://api.telegram.org/bot{token}/sendMessage" - for chat_id in chat_ids: - try: - requests.post( - url, - json={ - "chat_id": chat_id, - "text": text, - "disable_web_page_preview": True, - }, - timeout=10, - ).raise_for_status() - except Exception: - continue - - -def _send_alert_notifications(text: str) -> None: - _send_telegram_message(text) - - -class _Handler(BaseHTTPRequestHandler): - def do_GET(self) -> None: # noqa: N802 - if self.path.rstrip("/") == "/healthz": - self.send_response(200) - self.send_header("Content-Type", "application/json; charset=utf-8") - self.end_headers() - self.wfile.write(b'{"ok":true}') - return - self.send_error(404) - - def do_POST(self) -> None: # noqa: N802 - if self.path.rstrip("/") != "/alerts": - self.send_error(404) - return - length = int(self.headers.get("Content-Length", "0") or "0") - raw = self.rfile.read(length) if length > 0 else b"{}" - try: - payload = json.loads(raw.decode("utf-8")) - except Exception: - self.send_error(400, "invalid json") - return - if not isinstance(payload, dict): - self.send_error(400, "invalid payload") - return - _send_alert_notifications(_format_alerts(payload)) - self.send_response(200) - self.send_header("Content-Type", "application/json; charset=utf-8") - self.end_headers() - self.wfile.write(b'{"ok":true}') - - def log_message(self, format: str, *args: object) -> None: # noqa: A003 - return - - -def main() -> None: - port = 9099 - server = HTTPServer(("0.0.0.0", port), _Handler) - server.serve_forever() - - -if __name__ == "__main__": - main() diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py deleted file mode 100644 index a7c1f020..00000000 --- a/src/analysis/market_alert_engine.py +++ /dev/null @@ -1,1318 +0,0 @@ -""" -Rule-based weather alert engine for short-horizon trading signals. -""" - -from __future__ import annotations - -import math -import re -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]: - if v is None: - return None - try: - return float(v) - except Exception: - return None - - -def _to_unit_delta(celsius_delta: float, temp_symbol: str) -> float: - if "F" in (temp_symbol or "").upper(): - return celsius_delta * 9.0 / 5.0 - return celsius_delta - - -def _minute_of_day(hhmm: Optional[str]) -> Optional[int]: - if not hhmm or ":" not in str(hhmm): - return None - try: - hh, mm = str(hhmm).split(":")[:2] - h = int(hh) - m = int(mm) - if not (0 <= h <= 23 and 0 <= m <= 59): - return None - return h * 60 + m - except Exception: - return None - - -def _minutes_delta(newer_hhmm: Optional[str], older_hhmm: Optional[str]) -> Optional[int]: - newer = _minute_of_day(newer_hhmm) - older = _minute_of_day(older_hhmm) - if newer is None or older is None: - return None - d = newer - older - if d <= 0: - d += 24 * 60 - return d - - -def _angle_diff(a: float, b: float) -> float: - d = abs((a - b) % 360.0) - return min(d, 360.0 - d) - - -def _bearing_deg(lat1: float, lon1: float, lat2: float, lon2: float) -> float: - phi1 = math.radians(lat1) - phi2 = math.radians(lat2) - d_lon = math.radians(lon2 - lon1) - x = math.sin(d_lon) * math.cos(phi2) - y = math.cos(phi1) * math.sin(phi2) - math.sin(phi1) * math.cos(phi2) * math.cos(d_lon) - bearing = math.degrees(math.atan2(x, y)) - return (bearing + 360.0) % 360.0 - - -def _is_southerly(wdir: Optional[float]) -> bool: - if wdir is None: - return False - return 120.0 <= wdir <= 240.0 - - -def _calc_momentum_alert(city_weather: Dict[str, Any], temp_symbol: str) -> Dict[str, Any]: - recent = (city_weather.get("trend") or {}).get("recent") or [] - threshold_30m = _to_unit_delta(0.8, temp_symbol) - - if len(recent) < 2: - return { - "type": "momentum_spike", - "triggered": False, - "reason": "insufficient recent observations", - } - - newest = recent[0] - newest_temp = _sf(newest.get("temp")) - newest_time = newest.get("time") - if newest_temp is None: - return { - "type": "momentum_spike", - "triggered": False, - "reason": "latest observation missing temperature", - } - - anchor = None - anchor_dt = None - for row in recent[1:]: - dt = _minutes_delta(newest_time, row.get("time")) - if dt is None: - continue - # Prefer a point close to 30 minutes. - if 20 <= dt <= 45: - anchor = row - anchor_dt = dt - break - if anchor is None: - anchor = row - anchor_dt = dt - - if not anchor or not anchor_dt: - return { - "type": "momentum_spike", - "triggered": False, - "reason": "no usable time delta in recent observations", - } - - anchor_temp = _sf(anchor.get("temp")) - if anchor_temp is None: - return { - "type": "momentum_spike", - "triggered": False, - "reason": "anchor observation missing temperature", - } - - delta_temp = newest_temp - anchor_temp - slope_30m = delta_temp / anchor_dt * 30.0 - is_up = slope_30m > threshold_30m - is_down = slope_30m < -threshold_30m - - return { - "type": "momentum_spike", - "triggered": bool(is_up or is_down), - "direction": "up" if is_up else ("down" if is_down else "neutral"), - "newest_temp": round(newest_temp, 2), - "anchor_temp": round(anchor_temp, 2), - "delta_temp": round(delta_temp, 2), - "delta_minutes": anchor_dt, - "slope_30m": round(slope_30m, 2), - "threshold_30m": round(threshold_30m, 2), - } - - -def _pick_model_value(multi_model: Dict[str, Any], model_name: str) -> Optional[float]: - for k, v in (multi_model or {}).items(): - if str(k).strip().upper() == model_name.upper(): - return _sf(v) - return None - - -def _calc_forecast_breakthrough_alert(city_weather: Dict[str, Any], temp_symbol: str) -> Dict[str, Any]: - current_temp = _sf((city_weather.get("current") or {}).get("temp")) - if current_temp is None: - return { - "type": "forecast_breakthrough", - "triggered": False, - "reason": "current temperature unavailable", - } - - mm = city_weather.get("multi_model") or {} - mgm_high = _pick_model_value(mm, "MGM") - gfs_high = _pick_model_value(mm, "GFS") - ecmwf_high = _pick_model_value(mm, "ECMWF") - model_rows = [("MGM", mgm_high), ("GFS", gfs_high), ("ECMWF", ecmwf_high)] - available = [(k, v) for k, v in model_rows if v is not None] - - if not available: - return { - "type": "forecast_breakthrough", - "triggered": False, - "reason": "MGM/GFS/ECMWF highs are unavailable", - } - - baseline_name, baseline_val = max(available, key=lambda item: item[1]) - threshold = _to_unit_delta(0.2, temp_symbol) - margin = current_temp - baseline_val - triggered = margin > threshold and len(available) >= 2 - - return { - "type": "forecast_breakthrough", - "triggered": triggered, - "current_temp": round(current_temp, 2), - "model_highs": {k: v for k, v in available}, - "baseline_model": baseline_name, - "baseline_high": round(baseline_val, 2), - "margin": round(margin, 2), - "threshold": round(threshold, 2), - "model_coverage": f"{len(available)}/3", - } - - -def _pick_leading_station(city: str, nearby: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: - if not nearby: - return None - city_l = (city or "").lower() - - def _temp(row: Dict[str, Any]) -> float: - return _sf(row.get("temp")) or -999.0 - - if city_l == "ankara": - priority_rows = [] - for row in nearby: - name = str(row.get("name") or "").lower() - sid = str(row.get("istNo") or "").strip() - if sid == "17128" or "airport" in name or "17128" in name: - priority_rows.append(row) - if priority_rows: - return max(priority_rows, key=_temp) - - return max(nearby, key=_temp) - - -def _pick_ankara_center_station(nearby: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]: - if not nearby: - return None - - for row in nearby: - name = str(row.get("name") or "").strip().lower() - sid = str(row.get("istNo") or "").strip() - if sid == "17128": - return row - if name in {"airport (mgm/17128)", "ankara airport", "ankara esenboğa airport"}: - return row - 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 {} - current_temp = _sf(current.get("temp")) - wind_now = _sf(current.get("wind_dir")) - wind_speed = _sf(current.get("wind_speed_kt")) - - if current_temp is None: - return { - "type": "advection", - "triggered": False, - "reason": "current temperature unavailable", - } - - recent_obs = city_weather.get("metar_recent_obs") or [] - wind_prev = None - for obs in recent_obs[1:]: - w = _sf(obs.get("wdir")) - if w is not None: - wind_prev = w - break - - nearby = city_weather.get("mgm_nearby") or [] - lead_station = _pick_leading_station(city, nearby) - if not lead_station: - return { - "type": "advection", - "triggered": False, - "reason": "no nearby stations available", - } - - lead_temp = _sf(lead_station.get("temp")) - if lead_temp is None: - return { - "type": "advection", - "triggered": False, - "reason": "leading station temperature unavailable", - } - - lead_delta = lead_temp - current_temp - min_delta = _to_unit_delta(1.0, temp_symbol) - if city == "ankara": - # Ankara center station often leads airport by a bit less than 1C. - min_delta = _to_unit_delta(0.8, temp_symbol) - - turned_southerly = _is_southerly(wind_now) and (wind_prev is not None and not _is_southerly(wind_prev)) - warm_flow_now = _is_southerly(wind_now) and (wind_speed is None or wind_speed >= 6.0) - - alignment = None - aligned = True - st_lat = _sf(lead_station.get("lat")) - st_lon = _sf(lead_station.get("lon")) - city_lat = _sf(city_weather.get("lat")) - city_lon = _sf(city_weather.get("lon")) - if all(v is not None for v in (st_lat, st_lon, city_lat, city_lon, wind_now)): - station_to_city = _bearing_deg(st_lat, st_lon, city_lat, city_lon) - wind_to_dir = (wind_now + 180.0) % 360.0 # meteorological wind_dir is "from" - alignment = _angle_diff(station_to_city, wind_to_dir) - aligned = alignment <= 70.0 - - triggered = lead_delta >= min_delta and aligned and (turned_southerly or warm_flow_now) - - lead_minutes = None - if triggered: - if lead_delta >= _to_unit_delta(1.5, temp_symbol) and (alignment is None or alignment <= 45): - lead_minutes = "20-30" - else: - lead_minutes = "20-40" - - return { - "type": "advection", - "triggered": triggered, - "lead_station": { - "name": lead_station.get("name"), - "istNo": lead_station.get("istNo"), - "temp": round(lead_temp, 2), - }, - "lead_delta": round(lead_delta, 2), - "threshold_delta": round(min_delta, 2), - "wind_now": round(wind_now, 1) if wind_now is not None else None, - "wind_prev": round(wind_prev, 1) if wind_prev is not None else None, - "turned_southerly": turned_southerly, - "wind_alignment_deg": round(alignment, 1) if alignment is not None else None, - "lead_window_minutes": lead_minutes, - } - - -def _calc_peak_passed_guard(city_weather: Dict[str, Any], temp_symbol: str) -> Dict[str, Any]: - current = city_weather.get("current") or {} - current_temp = _sf(current.get("temp")) - max_so_far = _sf(current.get("max_so_far")) - max_temp_time = current.get("max_temp_time") - local_time = city_weather.get("local_time") - - if current_temp is None or max_so_far is None: - return {"suppressed": False, "reason": "missing current/max_so_far"} - - local_min = _minute_of_day(local_time) - peak_min = _minute_of_day(max_temp_time) - if local_min is None or peak_min is None: - return {"suppressed": False, "reason": "missing local_time/max_temp_time"} - - # Do not suppress in the morning; many cities still make their daily high later. - if local_min < (14 * 60 + 30): - return {"suppressed": False, "reason": "too early in local day"} - - if peak_min >= local_min: - return {"suppressed": False, "reason": "peak has not passed yet"} - - minutes_since_peak = local_min - peak_min - rollback = max_so_far - current_temp - rollback_threshold = _to_unit_delta(0.8, temp_symbol) - cooled_off = rollback >= rollback_threshold - suppressed = minutes_since_peak >= 45 and cooled_off - - return { - "suppressed": suppressed, - "reason": "late-day peak already passed" if suppressed else "cool-off threshold not met", - "current_temp": round(current_temp, 2), - "max_so_far": round(max_so_far, 2), - "max_temp_time": max_temp_time, - "local_time": local_time, - "minutes_since_peak": minutes_since_peak, - "rollback": round(rollback, 2), - "rollback_threshold": round(rollback_threshold, 2), - } - - -def _join_trigger_types_cn(rules: Dict[str, Dict[str, Any]]) -> str: - mapping = [ - ("ankara_center_deb_hit", "Center达到DEB"), - ("momentum_spike", "动量突变"), - ("forecast_breakthrough", "预测突破"), - ("advection", "暖平流"), - ] - parts = [name for key, name in mapping if rules.get(key, {}).get("triggered")] - return " + ".join(parts) - - -def _norm_probability(v: Any) -> Optional[float]: - n = _sf(v) - if n is None: - return None - if n > 1.0: - n = n / 100.0 - return max(0.0, min(1.0, n)) - - -def _fmt_percent(v: Any) -> str: - n = _norm_probability(v) - if n is None: - return "--" - return f"{n * 100:.1f}%" - - -def _fmt_cents(v: Any) -> str: - n = _norm_probability(v) - if n is None: - return "--" - cents = n * 100.0 - return f"{cents:.1f}c" - - -def _bucket_label(bucket: Any) -> Optional[str]: - if not isinstance(bucket, dict): - return None - direct = ( - str(bucket.get("label") or "").strip() - or str(bucket.get("bucket") or "").strip() - or str(bucket.get("range") or "").strip() - ) - if direct: - normalized = re.sub( - r"(? Optional[float]: - if not isinstance(row, dict): - return None - return _norm_probability(row.get("yes_buy")) - - -def _has_actionable_yes_buy_quote(row: Dict[str, Any]) -> bool: - quote = _row_yes_buy_prob(row) - # 0 usually means no actionable orderbook bid, not a tradable quote. - return quote is not None and quote > 0.0 - - -def _to_celsius(temp: Optional[float], temp_symbol: str) -> Optional[float]: - if temp is None: - return None - if "F" in (temp_symbol or "").upper(): - return (temp - 32.0) * 5.0 / 9.0 - return temp - - -def _extract_open_meteo_today_high_c(city_weather: Dict[str, Any]) -> Optional[float]: - forecast = city_weather.get("forecast") or {} - om_today = _sf(forecast.get("today_high")) - - if om_today is None: - om = city_weather.get("open-meteo") or {} - daily = om.get("daily") or {} - series = daily.get("temperature_2m_max") or [] - if isinstance(series, list) and series: - om_today = _sf(series[0]) - - if om_today is None: - return None - - temp_symbol = str(city_weather.get("temp_symbol") or "") - return _to_celsius(om_today, temp_symbol) - - -def _extract_multi_model_anchor_high_c( - city_weather: Dict[str, Any], -) -> Tuple[Optional[float], Optional[str]]: - multi_model = city_weather.get("multi_model") or {} - temp_symbol = str(city_weather.get("temp_symbol") or "") - if isinstance(multi_model, dict): - anchor_model: Optional[str] = None - anchor_high_c: Optional[float] = None - for model_name, raw_value in multi_model.items(): - value = _to_celsius(_sf(raw_value), temp_symbol) - if value is None: - continue - if anchor_high_c is None or value > anchor_high_c: - anchor_high_c = value - anchor_model = str(model_name or "").strip() or None - if anchor_high_c is not None: - return anchor_high_c, anchor_model - - # Fallback keeps behavior resilient when multi-model data is unexpectedly missing. - fallback_high_c = _extract_open_meteo_today_high_c(city_weather) - if fallback_high_c is not None: - return fallback_high_c, "Open-Meteo" - return None, None - - -def _bucket_value(row: Dict[str, Any]) -> Optional[float]: - for key in ("value", "temp"): - value = _sf(row.get(key)) - if value is not None: - return value - - label = str(row.get("label") or "").strip() - m = re.search(r"(-?\d+(?:\.\d+)?)", label) - if not m: - return None - return _sf(m.group(1)) - - -def _bucket_bounds(row: Dict[str, Any]) -> Optional[Tuple[Optional[float], Optional[float]]]: - value = _bucket_value(row) - if value is None: - return None - - label = str(row.get("label") or "").lower() - is_upper_tail = any(key in label for key in ("+", "or higher", "or above", "and above")) - is_lower_tail = any(key in label for key in ("<=", "or lower", "or below", "and below")) - - if is_upper_tail and not is_lower_tail: - return value, None - if is_lower_tail and not is_upper_tail: - return None, value - return value, value - - -def _distance_to_bucket(target: float, bounds: Tuple[Optional[float], Optional[float]]) -> float: - lower, upper = bounds - if lower is not None and target < lower: - return lower - target - if upper is not None and target > upper: - return target - upper - return 0.0 - - -def _pick_bucket_for_forecast( - rows: List[Dict[str, Any]], - forecast_settlement: Optional[int], - forecast_today_high_c: Optional[float], -) -> Optional[Dict[str, Any]]: - if not rows: - return None - - target = ( - float(forecast_settlement) - if forecast_settlement is not None - else forecast_today_high_c - ) - if target is None: - return None - - best_row: Optional[Dict[str, Any]] = None - best_distance: Optional[float] = None - best_has_quote = False - best_probability = -1.0 - best_rank = 10**9 - - for idx, row in enumerate(rows): - bounds = _bucket_bounds(row) - if not bounds: - continue - - distance = _distance_to_bucket(target, bounds) - has_quote = _has_actionable_yes_buy_quote(row) - probability = _norm_probability(row.get("probability")) - probability_rank = probability if probability is not None else -1.0 - - if best_row is None: - best_row = row - best_distance = distance - best_has_quote = has_quote - best_probability = probability_rank - best_rank = idx - continue - - assert best_distance is not None - if distance < best_distance: - best_row = row - best_distance = distance - best_has_quote = has_quote - best_probability = probability_rank - best_rank = idx - continue - - if abs(distance - best_distance) <= 1e-9: - if has_quote and not best_has_quote: - best_row = row - best_distance = distance - best_has_quote = has_quote - best_probability = probability_rank - best_rank = idx - elif has_quote == best_has_quote and probability_rank > best_probability: - best_row = row - best_distance = distance - best_has_quote = has_quote - best_probability = probability_rank - best_rank = idx - elif ( - has_quote == best_has_quote - and abs(probability_rank - best_probability) <= 1e-9 - and idx < best_rank - ): - best_row = row - best_distance = distance - best_has_quote = has_quote - best_probability = probability_rank - best_rank = idx - - return best_row - - -def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]: - scan = city_weather.get("market_scan") or {} - city = str(city_weather.get("name") or "").strip().lower() - if not isinstance(scan, dict): - return {"available": False} - if not scan.get("available"): - return {"available": False} - - yes_buy = _norm_probability(scan.get("yes_buy")) - yes_sell = _norm_probability(scan.get("yes_sell")) - market_prob = _norm_probability( - scan.get("market_price") - or ((scan.get("yes_token") or {}).get("implied_probability")) - ) - model_prob = _norm_probability(scan.get("model_probability")) - spread = None - if yes_buy is not None and yes_sell is not None: - spread = abs(yes_sell - yes_buy) - - top_bucket = None - top_bucket_rows: List[Dict[str, Any]] = [] - all_bucket_rows: List[Dict[str, Any]] = [] - source_buckets = scan.get("all_buckets") - if not isinstance(source_buckets, list) or not source_buckets: - source_buckets = scan.get("top_buckets") or [] - - if isinstance(source_buckets, list): - normalized = [] - for row in source_buckets: - if not isinstance(row, dict): - continue - p = _norm_probability(row.get("probability")) - if p is None: - continue - normalized.append((p, row)) - if normalized: - normalized.sort(key=lambda x: x[0], reverse=True) - top_bucket = normalized[0][1] - for p, row in normalized: - row_slug = str(row.get("slug") or "").strip() - row_market_url = f"https://polymarket.com/market/{row_slug}" if row_slug else None - all_bucket_rows.append( - { - "label": _bucket_label(row), - "probability": p, - "yes_buy": _norm_probability(row.get("yes_buy")), - "yes_sell": _norm_probability(row.get("yes_sell")), - "value": _sf(row.get("value") or row.get("temp")), - "slug": row_slug or None, - "market_url": row_market_url, - } - ) - top_bucket_rows = all_bucket_rows[:4] - - market_url = None - primary_market = scan.get("primary_market") or {} - if not isinstance(primary_market, dict): - primary_market = {} - websocket = scan.get("websocket") or {} - if isinstance(websocket, dict): - market_url = str(websocket.get("market_url") or "").strip() or None - if not market_url: - slug = str(primary_market.get("slug") or "").strip() - if slug: - market_url = f"https://polymarket.com/market/{slug}" - - anchor_today_high_c, anchor_model = _extract_multi_model_anchor_high_c(city_weather) - anchor_settlement = apply_city_settlement(city, anchor_today_high_c) - forecast_bucket = _pick_bucket_for_forecast( - rows=all_bucket_rows, - forecast_settlement=anchor_settlement, - forecast_today_high_c=anchor_today_high_c, - ) - forecast_market_url = None - if isinstance(forecast_bucket, dict): - forecast_market_url = str(forecast_bucket.get("market_url") or "").strip() or None - - return { - "available": True, - "selected_bucket": _bucket_label(scan.get("temperature_bucket")), - "top_bucket": _bucket_label(top_bucket) if isinstance(top_bucket, dict) else None, - "top_bucket_prob": _norm_probability( - top_bucket.get("probability") if isinstance(top_bucket, dict) else None - ), - "market_prob": market_prob, - "model_prob": model_prob, - "yes_buy": yes_buy, - "yes_sell": yes_sell, - "spread": spread, - "edge_percent": _sf(scan.get("edge_percent")), - "signal_label": scan.get("signal_label"), - "confidence": scan.get("confidence"), - "top_bucket_rows": top_bucket_rows, - "all_bucket_rows": all_bucket_rows, - "anchor_today_high_c": anchor_today_high_c, - "anchor_settlement": anchor_settlement, - "anchor_model": anchor_model, - # Backward-compatible aliases for existing consumers. - "open_meteo_today_high_c": anchor_today_high_c, - "open_meteo_settlement": anchor_settlement, - "forecast_bucket": forecast_bucket, - "selected_date": scan.get("selected_date"), - "selected_slug": scan.get("selected_slug"), - "primary_market": primary_market, - "market_active": primary_market.get("active"), - "market_closed": primary_market.get("closed"), - "market_accepting_orders": primary_market.get("accepting_orders"), - "market_tradable": primary_market.get("tradable"), - "market_tradable_reason": primary_market.get("tradable_reason"), - "market_ended_at_utc": primary_market.get("ended_at_utc"), - "primary_market_url": market_url, - "market_url": forecast_market_url or market_url, - } - - -def _build_advice_cn( - rules: Dict[str, Dict[str, Any]], - temp_symbol: str, - suppression: Optional[Dict[str, Any]] = None, -) -> str: - if (suppression or {}).get("suppressed"): - max_so_far = _sf((suppression or {}).get("max_so_far")) - max_temp_time = (suppression or {}).get("max_temp_time") - rollback = _sf((suppression or {}).get("rollback")) - if max_so_far is not None and max_temp_time and rollback is not None: - return ( - f"当地高温大概率已在 {max_temp_time} 前后兑现，" - f"较日内高点 {max_so_far:.1f}{temp_symbol} 已回落 {rollback:.1f}{temp_symbol}，暂停主动推送。" - ) - return "当地高温大概率已经兑现，当前进入回落阶段，暂停主动推送。" - - 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", {}) - - 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"): - d = _sf(momentum.get("slope_30m")) or 0.0 - if d > 0: - parts.append("短时升温斜率过快") - else: - parts.append("短时降温斜率过快") - if breakthrough.get("triggered"): - parts.append("实测已击穿主流模型上沿") - - if not parts: - return "当前未触发高优先级关键信号，继续观察实测与模型联动。" - return "，".join(parts) + "。" - - -_AIRPORT_ICAO_MAP = {"seoul": "RKSI", "busan": "RKPK", "tokyo": "RJTT", "ankara": "17128", "helsinki": "EFHK", "amsterdam": "EHAM", "istanbul": "17058", "paris": "LFPB", "hong kong": "HKO", "lau fau shan": "LFS", "taipei": "466920"} - - -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]], - map_url: Optional[str], - market_snapshot: Optional[Dict[str, Any]] = None, - suppression: Optional[Dict[str, Any]] = None, -) -> Dict[str, str]: - 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")) - local_time = str(city_weather.get("local_time") or "").strip() - obs_time = str(((city_weather.get("current") or {}).get("obs_time")) or "").strip() - center_deb = rules.get("ankara_center_deb_hit", {}) - momentum = rules.get("momentum_spike", {}) - advection = rules.get("advection", {}) - market_snapshot = market_snapshot or _extract_market_snapshot(city_weather) - - if current_temp is None: - return {"zh": "", "en": ""} - - suppressed = bool((suppression or {}).get("suppressed")) - has_active_trigger = any(rule.get("triggered") for rule in rules.values()) - if suppressed: - types_cn = "高温已过（暂停推送）" - else: - types_cn = _join_trigger_types_cn(rules) or "天气状态快照" - delta_temp = _sf(momentum.get("delta_temp")) - delta_min = momentum.get("delta_minutes") - 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: - icon = "🚀" if delta_temp > 0 else ("🧊" if delta_temp < 0 else "➖") - dyn += f" ({int(delta_min)}min 内 {delta_temp:+.1f}{temp_symbol}) {icon}" - - lead_line = "" - if advection.get("triggered"): - st_name = ((advection.get("lead_station") or {}).get("name")) or "nearby station" - lead_delta = _sf(advection.get("lead_delta")) - 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}" - - peak_line = "" - if suppressed: - max_so_far = _sf((suppression or {}).get("max_so_far")) - max_temp_time = (suppression or {}).get("max_temp_time") - rollback = _sf((suppression or {}).get("rollback")) - if max_so_far is not None and max_temp_time and rollback is not None: - peak_line = ( - f"高温状态：日内高点 {max_so_far:.1f}{temp_symbol} @ {max_temp_time}，" - f"当前已回落 {rollback:.1f}{temp_symbol}" - ) - - advice = _build_advice_cn(rules, temp_symbol, suppression=suppression) - final_map = map_url or "https://polyweather-pro.vercel.app/" - title_zh = "🚨 PolyWeather 市场提醒" if has_active_trigger else "📍 PolyWeather 状态快照" - title_en = "🚨 PolyWeather Market Alert" if has_active_trigger else "📍 PolyWeather Status" - - lines_zh = [ - f"{title_zh} [{city_name}]", - "", - f"类型：{types_cn}", - f"动态：{dyn}", - ] - if local_time or obs_time: - if local_time and obs_time: - lines_zh.append(f"时间：当地 {local_time} | 观测 {obs_time}") - elif local_time: - lines_zh.append(f"时间：当地 {local_time}") - else: - lines_zh.append(f"时间：观测 {obs_time}") - if center_deb_line: - lines_zh.append(center_deb_line) - if peak_line: - lines_zh.append(peak_line) - if lead_line: - lines_zh.append(lead_line) - if market_snapshot.get("available") and market_snapshot.get("top_bucket_rows"): - lines_zh.append("市场结算概率分布（Top4）：") - for row in (market_snapshot.get("top_bucket_rows") or [])[:4]: - label = row.get("label") or "--" - prob_text = _fmt_percent(row.get("probability")) - yes_buy_text = _fmt_cents(row.get("yes_buy")) - lines_zh.append(f"{label} {prob_text} | Yes: {yes_buy_text}") - if market_snapshot.get("available") and not market_snapshot.get("top_bucket_rows"): - market_edge = _sf(market_snapshot.get("edge_percent")) - market_edge_text = f"{market_edge:+.1f}%" if market_edge is not None else "--" - lines_zh.append( - "市场联动：同桶 " - f"模型 {_fmt_percent(market_snapshot.get('model_prob'))} vs " - f"市场 {_fmt_percent(market_snapshot.get('market_prob'))} | " - f"Yes {_fmt_cents(market_snapshot.get('yes_buy'))}/{_fmt_cents(market_snapshot.get('yes_sell'))} | " - f"点差 {_fmt_cents(market_snapshot.get('spread'))} | " - f"偏差 {market_edge_text} | " - f"信号 {market_snapshot.get('signal_label') or '--'}/{market_snapshot.get('confidence') or '--'}" - ) - if market_snapshot.get("top_bucket"): - lines_zh.append( - f"市场最热桶：{market_snapshot.get('top_bucket')} " - f"({_fmt_percent(market_snapshot.get('top_bucket_prob'))})" - ) - if market_snapshot.get("market_url"): - lines_zh.append(f"市场链接：{market_snapshot.get('market_url')}") - lines_zh.append(f"AI 建议：{advice}") - lines_zh.append(f"点击查看实时地图：{final_map}") - - type_en = [] - if rules.get("ankara_center_deb_hit", {}).get("triggered"): - type_en.append("Center Reached DEB") - if rules.get("momentum_spike", {}).get("triggered"): - type_en.append("Momentum Spike") - if rules.get("forecast_breakthrough", {}).get("triggered"): - type_en.append("Forecast Breakthrough") - if rules.get("advection", {}).get("triggered"): - type_en.append("Advection") - type_en_str = "Peak Passed (suppressed)" if suppressed else (" + ".join(type_en) or "Weather snapshot") - - lines_en = [ - f"{title_en} [{city_name}]", - "", - f"Type: {type_en_str}", - f"Now: {current_temp:.1f}{temp_symbol}", - ] - if local_time or obs_time: - if local_time and obs_time: - lines_en.append(f"Time: local {local_time} | observed {obs_time}") - elif local_time: - lines_en.append(f"Time: local {local_time}") - else: - lines_en.append(f"Time: observed {obs_time}") - 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 peak_line: - max_so_far = _sf((suppression or {}).get("max_so_far")) - max_temp_time = (suppression or {}).get("max_temp_time") - rollback = _sf((suppression or {}).get("rollback")) - if max_so_far is not None and max_temp_time and rollback is not None: - lines_en.append( - f"Peak state: intraday high {max_so_far:.1f}{temp_symbol} at {max_temp_time}, " - f"now off by {rollback:.1f}{temp_symbol}" - ) - if market_snapshot.get("available") and market_snapshot.get("top_bucket_rows"): - lines_en.append("Settlement distribution (Top4):") - for row in (market_snapshot.get("top_bucket_rows") or [])[:4]: - label = row.get("label") or "--" - prob_text = _fmt_percent(row.get("probability")) - yes_buy_text = _fmt_cents(row.get("yes_buy")) - lines_en.append(f"{label} {prob_text} | Yes: {yes_buy_text}") - if market_snapshot.get("available") and not market_snapshot.get("top_bucket_rows"): - market_edge = _sf(market_snapshot.get("edge_percent")) - market_edge_text = f"{market_edge:+.1f}%" if market_edge is not None else "--" - lines_en.append( - "Market: same-bucket " - f"model {_fmt_percent(market_snapshot.get('model_prob'))} vs " - f"market {_fmt_percent(market_snapshot.get('market_prob'))} | " - f"Yes {_fmt_cents(market_snapshot.get('yes_buy'))}/{_fmt_cents(market_snapshot.get('yes_sell'))} | " - f"spread {_fmt_cents(market_snapshot.get('spread'))} | " - f"edge {market_edge_text} | " - f"signal {market_snapshot.get('signal_label') or '--'}/{market_snapshot.get('confidence') or '--'}" - ) - if market_snapshot.get("top_bucket"): - lines_en.append( - f"Top market bucket: {market_snapshot.get('top_bucket')} " - f"({_fmt_percent(market_snapshot.get('top_bucket_prob'))})" - ) - if market_snapshot.get("market_url"): - lines_en.append(f"Market link: {market_snapshot.get('market_url')}") - lines_en.append(f"Action: {advice}") - lines_en.append(f"Map: {final_map}") - - return {"zh": "\n".join(lines_zh), "en": "\n".join(lines_en)} - - -def _select_rule_evidence(rule: Dict[str, Any], keys: List[str]) -> Dict[str, Any]: - out: Dict[str, Any] = {} - for key in keys: - if key in rule: - out[key] = rule.get(key) - return out - - -def _build_alert_evidence( - city_weather: Dict[str, Any], - rules: Dict[str, Dict[str, Any]], - triggered: List[Dict[str, Any]], - suppression: Dict[str, Any], - market_snapshot: Dict[str, Any], - temp_symbol: str, -) -> Dict[str, Any]: - current = city_weather.get("current") or {} - deb = city_weather.get("deb") or {} - - momentum = rules.get("momentum_spike") or {} - breakthrough = rules.get("forecast_breakthrough") or {} - advection = rules.get("advection") or {} - ankara_center = rules.get("ankara_center_deb_hit") or {} - - top_rows = [] - for row in (market_snapshot.get("top_bucket_rows") or [])[:4]: - if not isinstance(row, dict): - continue - top_rows.append( - { - "label": row.get("label"), - "probability": row.get("probability"), - "yes_buy": row.get("yes_buy"), - "yes_sell": row.get("yes_sell"), - "market_url": row.get("market_url"), - } - ) - - trigger_types = [row.get("type") for row in triggered if row.get("type")] - forecast_bucket = market_snapshot.get("forecast_bucket") or {} - - return { - "version": 1, - "city": city_weather.get("name"), - "generated_local_time": city_weather.get("local_time"), - "observed_at": current.get("obs_time"), - "temp_symbol": temp_symbol, - "inputs": { - "current_temp": _sf(current.get("temp")), - "deb_prediction": _sf(deb.get("prediction")), - "wu_settle": current.get("wu_settle"), - "obs_age_min": current.get("obs_age_min"), - }, - "trigger_summary": { - "trigger_count": len(trigger_types), - "trigger_types": trigger_types, - "suppressed": bool(suppression.get("suppressed")), - "suppression_reason": suppression.get("reason"), - "suppression_snapshot": _select_rule_evidence( - suppression, - [ - "minutes_since_peak", - "rollback", - "rollback_threshold", - "max_temp_time", - "max_so_far", - "current_temp", - ], - ), - }, - "rules": { - "momentum_spike": _select_rule_evidence( - momentum, - [ - "triggered", - "direction", - "delta_temp", - "delta_minutes", - "slope_30m", - "threshold_30m", - ], - ), - "forecast_breakthrough": _select_rule_evidence( - breakthrough, - [ - "triggered", - "baseline_model", - "baseline_high", - "current_temp", - "margin", - "threshold", - "model_coverage", - ], - ), - "advection": _select_rule_evidence( - advection, - [ - "triggered", - "lead_delta", - "threshold_delta", - "wind_now", - "wind_prev", - "turned_southerly", - "wind_alignment_deg", - "lead_window_minutes", - ], - ), - "ankara_center_deb_hit": _select_rule_evidence( - ankara_center, - [ - "triggered", - "deb_prediction", - "airport_temp", - "margin_vs_deb", - "center_lead_vs_airport", - ], - ), - }, - "market": { - "available": bool(market_snapshot.get("available")), - "low_yes_signal": { - "label": "removed", - "should_push": False, - "reason_cn": "mispricing signals have been removed", - }, - "market_prob": market_snapshot.get("market_prob"), - "model_prob": market_snapshot.get("model_prob"), - "edge_percent": market_snapshot.get("edge_percent"), - "yes_buy": market_snapshot.get("yes_buy"), - "yes_sell": market_snapshot.get("yes_sell"), - "spread": market_snapshot.get("spread"), - "signal_label": market_snapshot.get("signal_label"), - "confidence": market_snapshot.get("confidence"), - "top_bucket": market_snapshot.get("top_bucket"), - "top_bucket_prob": market_snapshot.get("top_bucket_prob"), - "anchor_today_high_c": market_snapshot.get("anchor_today_high_c"), - "anchor_settlement": market_snapshot.get("anchor_settlement"), - "anchor_model": market_snapshot.get("anchor_model"), - "open_meteo_today_high_c": market_snapshot.get("open_meteo_today_high_c"), - "open_meteo_settlement": market_snapshot.get("open_meteo_settlement"), - "forecast_bucket": { - "label": forecast_bucket.get("label"), - "probability": forecast_bucket.get("probability"), - "yes_buy": forecast_bucket.get("yes_buy"), - "yes_sell": forecast_bucket.get("yes_sell"), - "market_url": forecast_bucket.get("market_url"), - } - if isinstance(forecast_bucket, dict) - else None, - "top4": top_rows, - "market_url": market_snapshot.get("market_url"), - "primary_market_url": market_snapshot.get("primary_market_url"), - }, - } - - -def build_trading_alerts( - city_weather: Dict[str, Any], - map_url: Optional[str] = None, -) -> Dict[str, Any]: - """ - Build weather-driven trading alerts for paid Telegram delivery and web usage. - """ - temp_symbol = city_weather.get("temp_symbol", "°C") - city = city_weather.get("name", "") - now = datetime.now(timezone.utc).isoformat() - market_snapshot = _extract_market_snapshot(city_weather) - - rules: Dict[str, Dict[str, Any]] = { - "ankara_center_deb_hit": _calc_ankara_center_deb_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 = [ - { - "type": key, - **value, - } - for key, value in rules.items() - if value.get("triggered") - ] - suppression = _calc_peak_passed_guard(city_weather, temp_symbol) - if suppression.get("suppressed") and triggered: - suppression["raw_trigger_types"] = [alert.get("type") for alert in triggered if alert.get("type")] - for alert in triggered: - rule = rules.get(alert.get("type") or "") - if not rule: - continue - rule["raw_triggered"] = True - rule["triggered"] = False - rule["suppressed"] = True - rule["suppression_reason"] = suppression.get("reason") - triggered = [] - force_push = False - severity = "none" - else: - 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, - rules=rules, - map_url=map_url, - market_snapshot=market_snapshot, - suppression=suppression, - ) - evidence = _build_alert_evidence( - city_weather=city_weather, - rules=rules, - triggered=triggered, - suppression=suppression, - market_snapshot=market_snapshot, - temp_symbol=temp_symbol, - ) - - return { - "city": city, - "generated_at": now, - "temp_symbol": temp_symbol, - "severity": severity, - "trigger_count": len(triggered), - "rules": rules, - "market_snapshot": market_snapshot, - "suppression": suppression, - "triggered_alerts": triggered, - "evidence": evidence, - "telegram": telegram, - } diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 18bf018d..25b72ed3 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -83,7 +83,6 @@ 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(), @@ -152,37 +151,6 @@ class StartupCoordinator: details=details, ) - def _start_trade_alert_loop(self) -> LoopStatus: - enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True) - chat_ids = get_telegram_chat_ids_from_env() - interval = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 1800)) - cities_count = _parse_csv_count(os.getenv("TELEGRAM_ALERT_CITIES")) - details = { - "mode": "critical-alerts", - "interval_sec": interval, - "cities_count": cities_count, - "chat_targets": len(chat_ids), - "alert_cooldown_sec": max( - 60, - _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 21600), - ), - "min_trigger_count": max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 3)), - "min_severity": str(os.getenv("TELEGRAM_ALERT_MIN_SEVERITY") or "high").strip().lower(), - "daily_signal_cap": "none", - } - validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS" - return self._start_with_validation( - key="trade_alert_push", - label="市场监控推送", - configured_enabled=enabled, - details=details, - validation_error=validation_error, - starter=lambda: import_module("src.utils.telegram_push").start_trade_alert_push_loop( - self.bot, - self.config, - ), - ) - def _start_airport_high_freq_loop(self) -> LoopStatus: enabled = _env_bool("TELEGRAM_AIRPORT_PUSH_ENABLED", True) chat_ids = get_telegram_chat_ids_from_env() @@ -269,10 +237,9 @@ class StartupCoordinator: label="Polymarket 钱包异动监听（已停用）", configured_enabled=False, started=False, - reason="retired_replaced_by_market_monitor", + reason="retired", details={ - "replacement": "trade_alert_push", - "note": "wallet activity watcher retired in favor of market monitor digests and critical alerts", + "note": "wallet activity watcher retired", }, ) diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index b32512a5..552babd3 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -483,211 +483,6 @@ def _alert_signature(alert_payload: Dict[str, Any]) -> str: raw = json.dumps(signature_payload, sort_keys=True, ensure_ascii=True) return hashlib.sha1(raw.encode("utf-8")).hexdigest() - -def build_trade_alert_for_city( - city: str, - config: Dict[str, Any], - force_refresh: bool = False, - target_date: Optional[str] = None, -) -> Dict[str, Any]: - from web.app import _analyze, _build_city_detail_payload - from src.analysis.market_alert_engine import build_trading_alerts - - city_weather = _analyze(city, force_refresh=force_refresh) - try: - aggregate_detail = _build_city_detail_payload( - city_weather, - target_date=target_date, - ) - market_scan = aggregate_detail.get("market_scan") - if isinstance(market_scan, dict): - city_weather = {**city_weather, "market_scan": market_scan} - except Exception as exc: - logger.debug(f"market scan attach skipped city={city}: {exc}") - - resolved_target_date = target_date or city_weather.get("local_date") - if resolved_target_date: - datetime.strptime(resolved_target_date, "%Y-%m-%d") - - map_url = os.getenv("POLYWEATHER_MAP_URL") or "https://polyweather-pro.vercel.app/" - alert_payload = build_trading_alerts( - city_weather=city_weather, - map_url=map_url, - ) - alert_payload["target_date"] = resolved_target_date - return alert_payload - - -def _maybe_send_alert( - bot: Any, - chat_ids: List[str], - city: str, - alert_payload: Dict[str, Any], - state: Dict[str, Any], - cooldown_sec: int, - min_severity: str, - min_trigger_count: int, -) -> bool: - now_ts = int(time.time()) - last_by_city = state.setdefault("last_by_city", {}) - last_city = last_by_city.get(city) or {} - is_active = _severity_ok(alert_payload, min_severity, min_trigger_count) - if not _market_price_cap_ok(alert_payload): - is_active = False - message = ((alert_payload.get("telegram") or {}).get("zh") or "").strip() - - if not is_active or not message: - if last_city.get("active"): - last_by_city[city] = { - **last_city, - "active": False, - "cleared_ts": now_ts, - } - logger.info(f"market monitor disarmed city={city}") - return True - return False - - if not chat_ids: - return False - - signature = _alert_signature(alert_payload) - trigger_key = _trigger_type_key(alert_payload) - last_city_sig = last_city.get("signature") - last_city_key = str(last_city.get("trigger_key") or "") - last_city_ts = int(last_city.get("ts") or 0) - last_sig_ts = int((state.get("by_signature") or {}).get(signature) or 0) - last_city_active = bool(last_city.get("active")) - - if last_city_active and last_city_key == trigger_key and last_city_sig == signature: - return False - - if last_city_ts and now_ts - last_city_ts < cooldown_sec: - return False - if last_sig_ts and now_ts - last_sig_ts < cooldown_sec: - return False - - sent_count = 0 - for chat_id in chat_ids: - try: - bot.send_message(chat_id, message) - sent_count += 1 - except Exception as exc: - logger.warning("market monitor push failed city={} chat_id={} error={}", city, chat_id, exc) - if sent_count <= 0: - return False - - last_by_city[city] = { - "signature": signature, - "trigger_key": trigger_key, - "severity": alert_payload.get("severity"), - "ts": now_ts, - "active": True, - "evidence": alert_payload.get("evidence"), - } - state.setdefault("by_signature", {})[signature] = now_ts - logger.info( - f"market monitor pushed city={city} severity={alert_payload.get('severity')} " - f"trigger_count={alert_payload.get('trigger_count')} trigger_key={trigger_key} " - f"evidence={_evidence_brief(alert_payload)} chat_targets={sent_count}" - ) - return True - - -def _run_market_monitor_cycle( - bot: Any, - config: Dict[str, Any], - *, - chat_ids: List[str], - cities: List[str], - state: Dict[str, Any], - alert_cooldown_sec: int, - min_severity: str, - min_trigger_count: int, - sleep_between_cities_sec: float = 1.0, -) -> bool: - state_dirty = False - - for city in cities: - try: - alert_payload = build_trade_alert_for_city(city, config) - if _maybe_send_alert( - bot=bot, - chat_ids=chat_ids, - city=city, - alert_payload=alert_payload, - state=state, - cooldown_sec=alert_cooldown_sec, - min_severity=min_severity, - min_trigger_count=min_trigger_count, - ): - state_dirty = True - except Exception: - logger.exception(f"telegram market monitor loop failed for city={city}") - if sleep_between_cities_sec > 0: - time.sleep(sleep_between_cities_sec) - - return state_dirty - - -def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[threading.Thread]: - enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True) - chat_ids = get_telegram_chat_ids_from_env() - if not enabled: - logger.info("telegram market monitor loop disabled") - return None - if not chat_ids: - logger.warning("telegram market monitor loop skipped: TELEGRAM_CHAT_IDS is not set") - return None - - interval_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 1800)) - cities = _parse_city_list(os.getenv("TELEGRAM_ALERT_CITIES")) - state_path = _state_file() - alert_cooldown_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 21600)) - min_trigger_count = max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 3)) - min_severity = str(os.getenv("TELEGRAM_ALERT_MIN_SEVERITY") or "high").strip().lower() - if min_severity not in SEVERITY_RANK: - min_severity = "high" - def _runner() -> None: - try: - _save_state(state_path, _load_state(state_path)) - except Exception: - logger.exception(f"failed to initialize market monitor state path={state_path}") - logger.info( - f"telegram market monitor loop started mode=critical-alerts " - f"cities={len(cities)} interval={interval_sec}s chat_targets={len(chat_ids)} " - f"alert_cooldown={alert_cooldown_sec}s " - f"min_severity={min_severity} min_trigger_count={min_trigger_count} " - f"state_path={state_path}" - ) - while True: - cycle_started = time.time() - state = _load_state(state_path) - _cleanup_state(state, int(cycle_started)) - if _run_market_monitor_cycle( - bot=bot, - config=config, - chat_ids=chat_ids, - cities=cities, - state=state, - alert_cooldown_sec=alert_cooldown_sec, - min_severity=min_severity, - min_trigger_count=min_trigger_count, - ): - _save_state(state_path, state) - - elapsed = time.time() - cycle_started - sleep_sec = max(5, interval_sec - int(elapsed)) - time.sleep(sleep_sec) - - thread = threading.Thread( - target=_runner, - name="telegram-market-monitor-pusher", - daemon=True, - ) - thread.start() - return thread - - # ── high-freq airport push loop ── HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam", "istanbul", "paris", "hong kong", "lau fau shan", "taipei"}