From d6434bf174d2e9e129f46586c817a980146301d0 Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Wed, 11 Mar 2026 01:56:22 +0800 Subject: [PATCH] feat: Add weather alert engine, Polymarket data reader, and Telegram notification utility. --- src/analysis/market_alert_engine.py | 245 ++++++++++++++++++++- src/data_collection/polymarket_readonly.py | 14 +- src/utils/telegram_push.py | 80 +++++++ 3 files changed, 327 insertions(+), 12 deletions(-) diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index 2ac3cdcf..de68f123 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -7,7 +7,9 @@ from __future__ import annotations import math import re from datetime import datetime, timezone -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple + +from src.analysis.settlement_rounding import wu_round def _sf(v: Any) -> Optional[float]: @@ -471,6 +473,130 @@ def _bucket_label(bucket: Any) -> Optional[str]: return None +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 _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_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) + 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_probability = probability_rank + best_rank = idx + continue + + assert best_distance is not None + if distance < best_distance: + best_row = row + best_distance = distance + best_probability = probability_rank + best_rank = idx + continue + + if abs(distance - best_distance) <= 1e-9: + if probability_rank > best_probability: + best_row = row + best_distance = distance + best_probability = probability_rank + best_rank = idx + elif abs(probability_rank - best_probability) <= 1e-9 and idx < best_rank: + best_row = row + best_distance = distance + 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 {} if not isinstance(scan, dict): @@ -491,10 +617,14 @@ def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]: top_bucket = None top_bucket_rows: List[Dict[str, Any]] = [] - top_buckets = scan.get("top_buckets") or [] - if isinstance(top_buckets, list): + 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 top_buckets: + for row in source_buckets: if not isinstance(row, dict): continue p = _norm_probability(row.get("probability")) @@ -504,15 +634,21 @@ def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]: if normalized: normalized.sort(key=lambda x: x[0], reverse=True) top_bucket = normalized[0][1] - for p, row in normalized[:4]: - top_bucket_rows.append( + 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 websocket = scan.get("websocket") or {} @@ -525,6 +661,17 @@ def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]: if slug: market_url = f"https://polymarket.com/market/{slug}" + open_meteo_today_high_c = _extract_open_meteo_today_high_c(city_weather) + open_meteo_settlement = wu_round(open_meteo_today_high_c) + forecast_bucket = _pick_bucket_for_forecast( + rows=all_bucket_rows, + forecast_settlement=open_meteo_settlement, + forecast_today_high_c=open_meteo_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")), @@ -541,7 +688,12 @@ def _extract_market_snapshot(city_weather: Dict[str, Any]) -> Dict[str, Any]: "signal_label": scan.get("signal_label"), "confidence": scan.get("confidence"), "top_bucket_rows": top_bucket_rows, - "market_url": market_url, + "all_bucket_rows": all_bucket_rows, + "open_meteo_today_high_c": open_meteo_today_high_c, + "open_meteo_settlement": open_meteo_settlement, + "forecast_bucket": forecast_bucket, + "primary_market_url": market_url, + "market_url": forecast_market_url or market_url, } @@ -786,6 +938,81 @@ def _build_telegram_messages( return {"zh": "\n".join(lines_zh), "en": "\n".join(lines_en)} +def _build_telegram_messages_mispricing( + city_weather: Dict[str, Any], + rules: Dict[str, Dict[str, Any]], + market_snapshot: Optional[Dict[str, Any]] = None, +) -> Dict[str, str]: + temp_symbol = str(city_weather.get("temp_symbol") or "°C") + city_name = city_weather.get("display_name") or city_weather.get("name", "").title() + current = city_weather.get("current") or {} + current_temp = _sf(current.get("temp")) + if current_temp is None: + return {"zh": "", "en": ""} + + snapshot = market_snapshot or _extract_market_snapshot(city_weather) + momentum = rules.get("momentum_spike", {}) + local_time = str(city_weather.get("local_time") or "").strip() + obs_time = str(current.get("obs_time") or "").strip() + + delta_temp = _sf(momentum.get("delta_temp")) + delta_min = momentum.get("delta_minutes") + momentum_emoji = "➡️" + if delta_temp is not None: + momentum_emoji = "🚀" if delta_temp > 0 else ("📉" if delta_temp < 0 else "➡️") + + dynamic_text = f"实测 {current_temp:.1f}{temp_symbol}" + if delta_temp is not None and delta_min is not None: + dynamic_text = ( + f"实测 {current_temp:.1f}{temp_symbol} " + f"({int(delta_min)}min 内 {delta_temp:+.1f}{temp_symbol}) {momentum_emoji}" + ) + + om_high_c = _sf(snapshot.get("open_meteo_today_high_c")) + om_settle = snapshot.get("open_meteo_settlement") + forecast_bucket = snapshot.get("forecast_bucket") or {} + match_bucket_label = str(forecast_bucket.get("label") or "--").strip() or "--" + match_bucket_yes = _fmt_cents(forecast_bucket.get("yes_buy")) + market_url = str( + snapshot.get("market_url") + or snapshot.get("primary_market_url") + or "" + ).strip() + + lines_zh = [f"🚨 PolyWeather 错价雷达 [{city_name}]"] + lines_zh.append("") + if om_high_c is not None and om_settle is not None: + lines_zh.append( + f"基准：Open-Meteo 今日高温 {om_high_c:.1f}C（结算参考 {om_settle}C）" + ) + else: + lines_zh.append("基准：Open-Meteo 今日高温 --（结算参考 --）") + lines_zh.append(f"命中桶：{match_bucket_label} | Yes: {match_bucket_yes}") + lines_zh.append("触发：该桶 Yes 价格 < 10c，疑似低估") + lines_zh.append("") + lines_zh.append(f"动态：{dynamic_text}") + 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}") + lines_zh.append("") + if market_url: + lines_zh.append(f"市场链接：{market_url}") + + lines_en = [ + f"🚨 PolyWeather Mispricing Radar [{city_name}]", + "", + f"Now: {dynamic_text}", + ] + if market_url: + lines_en.append(f"Market link: {market_url}") + + return {"zh": "\n".join(lines_zh), "en": "\n".join(lines_en)} + + def build_trading_alerts( city_weather: Dict[str, Any], map_url: Optional[str] = None, @@ -833,12 +1060,10 @@ def build_trading_alerts( if force_push and severity == "none": severity = "medium" - telegram = _build_telegram_messages( + telegram = _build_telegram_messages_mispricing( city_weather=city_weather, rules=rules, - map_url=map_url, market_snapshot=market_snapshot, - suppression=suppression, ) return { diff --git a/src/data_collection/polymarket_readonly.py b/src/data_collection/polymarket_readonly.py index c371e4b9..82491b1a 100644 --- a/src/data_collection/polymarket_readonly.py +++ b/src/data_collection/polymarket_readonly.py @@ -486,12 +486,21 @@ class PolymarketReadOnlyLayer: signal_label, confidence = self._derive_signal(edge_percent, liquidity) - top_buckets = self._build_top_temperature_buckets( + top_bucket_limit = max( + 1, + _safe_int(os.getenv("POLYMARKET_TOP_BUCKET_LIMIT", "4"), 4), + ) + all_bucket_limit = max( + top_bucket_limit, + _safe_int(os.getenv("POLYMARKET_ALL_BUCKET_LIMIT", "24"), 24), + ) + all_buckets = self._build_top_temperature_buckets( city_key=city_key, target_date=date_str, primary_market=market, - limit=4, + limit=all_bucket_limit, ) + top_buckets = list(all_buckets[:top_bucket_limit]) yes_payload = { "outcome": yes_token.get("outcome") or "Yes", @@ -560,6 +569,7 @@ class PolymarketReadOnlyLayer: "volume": volume, "sparkline": sparkline_values, "top_buckets": top_buckets, + "all_buckets": all_buckets, "websocket": { "market_url": market_url, "asset_ids": [ diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 0b21cc6d..4a8a8edd 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -36,6 +36,28 @@ def _env_int(name: str, default: int) -> int: return default +def _env_float(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return default + try: + return float(raw) + except Exception: + return default + + +def _norm_prob(v: Any) -> Optional[float]: + if v is None: + return None + try: + n = float(v) + except Exception: + return None + if n > 1.0: + n = n / 100.0 + return max(0.0, min(1.0, n)) + + def _parse_city_list(raw: Optional[str]) -> List[str]: if not raw: return list(CITY_REGISTRY.keys()) @@ -111,6 +133,58 @@ def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_c return SEVERITY_RANK.get(severity, 0) >= SEVERITY_RANK.get(min_severity, 0) +def _market_price_cap_ok(alert_payload: Dict[str, Any], max_yes_buy: float) -> bool: + if max_yes_buy >= 1.0: + return True + + market = alert_payload.get("market_snapshot") or {} + if not isinstance(market, dict) or not market.get("available"): + return True + + # Prefer the market bucket that maps to Open-Meteo forecast settlement. + forecast_bucket = market.get("forecast_bucket") or {} + yes_buy = None + bucket_label = None + if isinstance(forecast_bucket, dict): + yes_buy = _norm_prob(forecast_bucket.get("yes_buy")) + bucket_label = str(forecast_bucket.get("label") or "").strip() or None + + # Backward-compatible fallback. + if yes_buy is None: + yes_buy = _norm_prob(market.get("yes_buy")) + if not bucket_label: + bucket_label = str(market.get("selected_bucket") or "").strip() or None + + if yes_buy is None: + # Fallback to first bucket with valid yes_buy if aggregate field is missing. + top_rows = market.get("top_bucket_rows") or [] + if isinstance(top_rows, list): + for row in top_rows: + if not isinstance(row, dict): + continue + yes_buy = _norm_prob(row.get("yes_buy")) + if yes_buy is not None: + if not bucket_label: + bucket_label = str(row.get("label") or "").strip() or None + break + + if yes_buy is None: + return True + + if yes_buy >= max_yes_buy: + logger.info( + "trade alert skipped by mispricing cap city={} bucket={} om_settle={} yes_buy={} cap={}".format( + alert_payload.get("city"), + bucket_label or "--", + market.get("open_meteo_settlement"), + round(yes_buy, 4), + round(max_yes_buy, 4), + ) + ) + return False + return True + + def _trigger_type_key(alert_payload: Dict[str, Any]) -> str: trigger_types = sorted( str(alert.get("type") or "").strip() @@ -218,6 +292,12 @@ def _maybe_send_alert( 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) + max_yes_buy = max( + 0.0, + min(1.0, _env_float("TELEGRAM_ALERT_MISPRICING_MAX_YES_BUY", 0.10)), + ) + if not _market_price_cap_ok(alert_payload, max_yes_buy): + is_active = False message = ((alert_payload.get("telegram") or {}).get("zh") or "").strip() if not is_active or not message: