diff --git a/docs/CONFIGURATION_ZH.md b/docs/CONFIGURATION_ZH.md index 24fd3384..c4001a08 100644 --- a/docs/CONFIGURATION_ZH.md +++ b/docs/CONFIGURATION_ZH.md @@ -287,7 +287,7 @@ POLYWEATHER_SCAN_CITY_AI_MODEL=mimo-v2.5-pro - `docker-compose.yml` 会把这个目录同时挂载到容器内的 `/var/lib/polyweather` 和 `/app/data`，兼容现有缓存与 SQLite 路径。 - `POLYWEATHER_STATE_STORAGE_MODE` 当前线上推荐直接使用 `sqlite`。 - `POLYWEATHER_PAYMENT_RPC_URLS` 支持逗号分隔多个 RPC；如果暂时只用单 RPC，也可以继续只配 `POLYWEATHER_PAYMENT_RPC_URL`。 -- 机器人市场监控当前以 `关注清单` 为主，按固定间隔主动推送。 +- 机器人市场监控包含 `关键提醒` 与 `关注清单`：关键提醒逐城判断并受冷却控制，关注清单每轮先扫描完整城市列表，再按全局 Top N 推送；同一轮已经触发关键提醒的城市不会重复出现在关注清单里。 - `POLYWEATHER_SCAN_AI_*` 走 OpenAI-compatible `/chat/completions`；当前临时默认 MiMo，可用 `POLYWEATHER_SCAN_AI_BASE_URL` 与 `POLYWEATHER_SCAN_AI_MODEL` 随时切回其他兼容 provider。 - `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC` 表示主动推送间隔，默认 `1800` 秒（30 分钟）。 - `POLYMARKET_WALLET_ACTIVITY_ENABLED` 已退役，保留为 `false` 即可，不建议再启用钱包异动监听。 @@ -354,7 +354,7 @@ POLYMARKET_WALLET_ACTIVITY_ENABLED=false 说明： - `TELEGRAM_ALERT_MISPRICING_ONLY=true` 表示关键提醒优先围绕错价/市场触发，不把机器人做成泛通知器。 -- `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800` 表示频道每 30 分钟主动推送一轮机会清单。 +- `TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800` 表示频道每 30 分钟主动推送一轮全局机会清单；每轮会先扫描完整 `TELEGRAM_ALERT_CITIES`，再选 Top N。 - `TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5` 建议先保持较小，避免机器人一次推太多城市。 - `POLYMARKET_WALLET_ACTIVITY_ENABLED=false` 表示停用旧的钱包异动监听，统一收敛到市场监控。 diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 3dfb7b3f..901129bd 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -157,10 +157,21 @@ class StartupCoordinator: interval = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300)) cities_count = _parse_csv_count(os.getenv("TELEGRAM_ALERT_CITIES")) details = { - "mode": "focus-digest-only", + "mode": "critical-alerts-and-focus-digest", "interval_sec": interval, "cities_count": cities_count, "chat_targets": len(chat_ids), + "alert_cooldown_sec": max( + 60, + _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 1800), + ), + "mispricing_interval_sec": max( + max(60, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 1800)), + _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 7200), + ), + "min_trigger_count": max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 2)), + "min_severity": str(os.getenv("TELEGRAM_ALERT_MIN_SEVERITY") or "medium").strip().lower(), + "mispricing_only": _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True), "focus_digest_enabled": _env_bool("TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED", True), "focus_digest_interval_sec": max( 300, diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 12a0915e..b7c2483d 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -337,9 +337,9 @@ def _focus_push_window_config() -> Dict[str, int]: def _focus_push_window_label() -> str: config = _focus_push_window_config() return ( - f"当地 {config['start_local_hour']:02d}:00 后，" - f"峰值前 {_format_minutes_window(config['before_peak_min'])} / " - f"峰值后 {_format_minutes_window(config['after_peak_min'])} 内" + f"当地 {config['start_local_hour']:02d}:00 后；" + f"有日内高点参考时，保留高点前 {_format_minutes_window(config['before_peak_min'])} / " + f"高点后 {_format_minutes_window(config['after_peak_min'])} 内" ) @@ -464,6 +464,39 @@ def _focus_trigger_summary(alert_payload: Dict[str, Any]) -> str: return _join_trigger_types_cn_local(rules) or "市场与天气分歧待观察" +def _market_signal_brief(snapshot: Dict[str, Any]) -> str: + if not isinstance(snapshot, dict): + return "" + + signal = str(snapshot.get("signal_label") or "").strip().upper() + signal_label = { + "BUY YES": "方向 BUY YES", + "BUY NO": "方向 BUY NO", + "MONITOR": "方向 MONITOR", + }.get(signal, f"方向 {signal}" if signal else "") + + edge = _safe_float(snapshot.get("edge_percent")) + confidence = str(snapshot.get("confidence") or "").strip() + forecast_bucket = snapshot.get("forecast_bucket") or {} + if not isinstance(forecast_bucket, dict): + forecast_bucket = {} + yes_buy = _fmt_cents(forecast_bucket.get("yes_buy")) + yes_sell = _fmt_cents(forecast_bucket.get("yes_sell")) + + parts: List[str] = [] + if signal_label: + parts.append(signal_label) + if edge is not None: + parts.append(f"edge {edge:+.1f}%") + if confidence: + parts.append(f"置信 {confidence}") + if yes_buy: + parts.append(f"Yes买 {yes_buy}") + if yes_sell: + parts.append(f"Yes卖 {yes_sell}") + return " | ".join(parts) + + def _shortlist_focus_payloads( payloads: List[Dict[str, Any]], *, @@ -538,6 +571,9 @@ def _build_focus_digest_message( score = _market_monitor_score(payload) lines.append(f"{idx}. {city_name} | {_priority_label(score)}") lines.append(" " + f"关注桶 {bucket}") + signal_brief = _market_signal_brief(snapshot) + if signal_brief: + lines.append(" " + signal_brief) local_time = str(peak_context.get("local_time") or "").strip() peak_time = str(peak_context.get("peak_time") or "").strip() window_label = str(peak_context.get("window_label") or "").strip() @@ -546,7 +582,7 @@ def _build_focus_digest_message( if local_time: context_parts.append(f"当地 {local_time}") if peak_time: - context_parts.append(f"峰值参考 {peak_time}") + context_parts.append(f"日内高点参考 {peak_time}") if window_label: context_parts.append(window_label) lines.append(" " + " | ".join(context_parts)) @@ -1027,6 +1063,8 @@ 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) + if mispricing_only and not _is_market_mispricing_signal(alert_payload): + is_active = False if not _market_price_cap_ok( alert_payload, require_actionable_quote=mispricing_only, @@ -1091,6 +1129,95 @@ def _maybe_send_alert( return True +def _is_market_mispricing_signal(alert_payload: Dict[str, Any]) -> bool: + market = alert_payload.get("market_snapshot") or {} + if not isinstance(market, dict) or not market.get("available"): + return False + + evidence = alert_payload.get("evidence") or {} + evidence_market = evidence.get("market") if isinstance(evidence, dict) else {} + low_yes_signal = ( + evidence_market.get("low_yes_signal") + if isinstance(evidence_market, dict) + else None + ) + if isinstance(low_yes_signal, dict) and bool(low_yes_signal.get("should_push")): + return True + + signal = str(market.get("signal_label") or "").strip().upper() + if signal in {"BUY YES", "BUY NO"}: + return True + return False + + +def _run_market_monitor_cycle( + bot: Any, + config: Dict[str, Any], + *, + chat_ids: List[str], + cities: List[str], + state: Dict[str, Any], + focus_digest_enabled: bool, + focus_digest_interval_sec: int, + focus_digest_top_n: int, + alert_cooldown_sec: int, + mispricing_interval_sec: int, + min_severity: str, + min_trigger_count: int, + mispricing_only: bool, + sleep_between_cities_sec: float = 1.0, +) -> bool: + state_dirty = False + cycle_payloads: List[Dict[str, Any]] = [] + critical_pushed_cities: set[str] = set() + + for city in cities: + try: + alert_payload = build_trade_alert_for_city(city, config) + cycle_payloads.append(alert_payload) + is_mispricing = _is_market_mispricing_signal(alert_payload) + cooldown_sec = mispricing_interval_sec if is_mispricing else alert_cooldown_sec + if _maybe_send_alert( + bot=bot, + chat_ids=chat_ids, + city=city, + alert_payload=alert_payload, + state=state, + cooldown_sec=cooldown_sec, + min_severity=min_severity, + min_trigger_count=min_trigger_count, + mispricing_only=mispricing_only, + ): + state_dirty = True + critical_pushed_cities.add(str(city).strip().lower()) + 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) + + if focus_digest_enabled: + try: + digest_payloads = [ + payload + for payload in cycle_payloads + if str(payload.get("city") or "").strip().lower() + not in critical_pushed_cities + ] + if _maybe_send_focus_digest( + bot=bot, + chat_ids=chat_ids, + payloads=digest_payloads, + state=state, + digest_interval_sec=focus_digest_interval_sec, + top_n=focus_digest_top_n, + ): + state_dirty = True + except Exception: + logger.exception("failed to push market focus digest") + + 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() @@ -1104,65 +1231,56 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th interval_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300)) 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", 1800)) + mispricing_interval_sec = max( + alert_cooldown_sec, + _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 7200), + ) + min_trigger_count = max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 2)) + min_severity = str(os.getenv("TELEGRAM_ALERT_MIN_SEVERITY") or "medium").strip().lower() + if min_severity not in SEVERITY_RANK: + min_severity = "medium" + mispricing_only = _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True) focus_digest_enabled = _env_bool("TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED", True) focus_digest_top_n = max(3, min(8, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N", 5))) focus_digest_interval_sec = max( 300, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 1800), ) - focus_digest_batch_size = max(4, min(12, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_BATCH_SIZE", 8))) 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=focus-digest-only " + f"telegram market monitor loop started mode=critical-alerts-and-focus-digest " f"cities={len(cities)} interval={interval_sec}s chat_targets={len(chat_ids)} " + f"alert_cooldown={alert_cooldown_sec}s mispricing_interval={mispricing_interval_sec}s " + f"min_severity={min_severity} min_trigger_count={min_trigger_count} " + f"mispricing_only={mispricing_only} " f"focus_digest_enabled={focus_digest_enabled} focus_interval={focus_digest_interval_sec}s " - f"focus_batch_size={focus_digest_batch_size} " f"state_path={state_path}" ) while True: cycle_started = time.time() state = _load_state(state_path) _cleanup_state(state, int(cycle_started)) - cycle_payloads: List[Dict[str, Any]] = [] - - for city in cities: - try: - alert_payload = build_trade_alert_for_city(city, config) - cycle_payloads.append(alert_payload) - except Exception: - logger.exception(f"telegram market monitor loop failed for city={city}") - if focus_digest_enabled and cycle_payloads and len(cycle_payloads) % focus_digest_batch_size == 0: - try: - if _maybe_send_focus_digest( - bot=bot, - chat_ids=chat_ids, - payloads=cycle_payloads, - state=state, - digest_interval_sec=focus_digest_interval_sec, - top_n=focus_digest_top_n, - ): - _save_state(state_path, state) - except Exception: - logger.exception("failed to push market focus digest mid-cycle") - time.sleep(1) - - if focus_digest_enabled: - try: - if _maybe_send_focus_digest( - bot=bot, - chat_ids=chat_ids, - payloads=cycle_payloads, - state=state, - digest_interval_sec=focus_digest_interval_sec, - top_n=focus_digest_top_n, - ): - _save_state(state_path, state) - except Exception: - logger.exception("failed to push market focus digest") + if _run_market_monitor_cycle( + bot=bot, + config=config, + chat_ids=chat_ids, + cities=cities, + state=state, + focus_digest_enabled=focus_digest_enabled, + focus_digest_interval_sec=focus_digest_interval_sec, + focus_digest_top_n=focus_digest_top_n, + alert_cooldown_sec=alert_cooldown_sec, + mispricing_interval_sec=mispricing_interval_sec, + min_severity=min_severity, + min_trigger_count=min_trigger_count, + mispricing_only=mispricing_only, + ): + _save_state(state_path, state) elapsed = time.time() - cycle_started sleep_sec = max(5, interval_sec - int(elapsed)) diff --git a/tests/test_market_alert_engine.py b/tests/test_market_alert_engine.py index 9a90d22a..a01202d3 100644 --- a/tests/test_market_alert_engine.py +++ b/tests/test_market_alert_engine.py @@ -1,6 +1,7 @@ from src.analysis.market_alert_engine import build_trading_alerts from src.utils.telegram_push import ( _build_focus_digest_message, + _run_market_monitor_cycle, _shortlist_focus_payloads, build_market_monitor_digest, ) @@ -201,23 +202,57 @@ def test_market_monitor_digest_skips_non_tradable_market(monkeypatch): assert digest == "ℹ️ 当前没有可用的市场监控摘要。" -def _focus_payload(*, local_time: str = "14:00", peak_time: str = "12:00"): +class _DigestBot: + def __init__(self): + self.messages = [] + + def send_message(self, chat_id, text, **kwargs): + self.messages.append( + { + "chat_id": chat_id, + "text": text, + "kwargs": kwargs, + } + ) + + +def _focus_payload( + *, + city: str = "madrid", + local_time: str = "14:00", + peak_time: str = "12:00", + severity: str = "medium", + trigger_count: int = 1, + edge_percent: float = 5.0, + signal_label: str = "MONITOR", + yes_buy: float = 0.18, +): return { - "city": "madrid", - "severity": "medium", - "trigger_count": 1, - "rules": {}, + "city": city, + "severity": severity, + "trigger_count": trigger_count, + "rules": { + "momentum_spike": { + "triggered": trigger_count > 0, + }, + }, "market_snapshot": { "available": True, "forecast_bucket": { "label": "30°C", "value": 30, - "yes_buy": 0.18, + "yes_buy": yes_buy, + "yes_sell": min(0.99, yes_buy + 0.02), "market_url": "https://example.com/market", }, "market_url": "https://example.com/market", "confidence": "medium", - "edge_percent": 5.0, + "edge_percent": edge_percent, + "signal_label": signal_label, + "market_active": True, + "market_closed": False, + "market_accepting_orders": True, + "market_tradable": True, }, "evidence": { "generated_local_time": local_time, @@ -226,10 +261,26 @@ def _focus_payload(*, local_time: str = "14:00", peak_time: str = "12:00"): "deb_prediction": 30.2, }, "trigger_summary": { + "trigger_types": ["momentum_spike"] if trigger_count > 0 else [], "suppression_snapshot": { "max_temp_time": peak_time, }, }, + "market": { + "low_yes_signal": { + "should_push": signal_label in {"BUY YES", "BUY NO"}, + }, + }, + }, + "triggered_alerts": [ + { + "type": "momentum_spike", + } + ] + if trigger_count > 0 + else [], + "telegram": { + "zh": f"CRITICAL {city}", }, } @@ -258,3 +309,150 @@ def test_focus_digest_footer_explains_no_daily_signal_cap(monkeypatch): assert "没有每日信号次数上限" in message assert "观察窗口" in message + + +def test_focus_digest_message_shows_market_direction_and_signed_edge(): + message = _build_focus_digest_message( + [ + _focus_payload( + city="new york", + signal_label="BUY NO", + edge_percent=-9.25, + ) + ], + slot_label="白天关注", + top_n=5, + ) + + assert "方向 BUY NO" in message + assert "edge -9.2%" in message + + +def test_market_monitor_cycle_sends_digest_after_full_city_scan(monkeypatch): + payloads = { + "early a": _focus_payload(city="early a", edge_percent=4.0, yes_buy=0.2), + "early b": _focus_payload(city="early b", edge_percent=4.0, yes_buy=0.2), + "late best": _focus_payload( + city="late best", + severity="high", + trigger_count=3, + edge_percent=20.0, + signal_label="BUY YES", + yes_buy=0.05, + ), + } + + monkeypatch.setattr( + "src.utils.telegram_push.build_trade_alert_for_city", + lambda city, config: payloads[city], + ) + monkeypatch.setenv("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", "1800") + bot = _DigestBot() + state = {} + + dirty = _run_market_monitor_cycle( + bot=bot, + config={}, + chat_ids=["chat"], + cities=["early a", "early b", "late best"], + state=state, + focus_digest_enabled=True, + focus_digest_interval_sec=1800, + focus_digest_top_n=3, + alert_cooldown_sec=1800, + mispricing_interval_sec=7200, + min_severity="high", + min_trigger_count=99, + mispricing_only=True, + sleep_between_cities_sec=0, + ) + + assert dirty is True + assert len(bot.messages) == 1 + assert "Late Best" in bot.messages[0]["text"] + assert "Early A" in bot.messages[0]["text"] + assert state.get("last_focus_digest_ts") + + +def test_market_monitor_cycle_restores_critical_alert_push(monkeypatch): + payload = _focus_payload( + city="ankara", + severity="high", + trigger_count=2, + signal_label="BUY YES", + yes_buy=0.05, + edge_percent=12.0, + ) + monkeypatch.setattr( + "src.utils.telegram_push.build_trade_alert_for_city", + lambda city, config: payload, + ) + bot = _DigestBot() + state = {} + + dirty = _run_market_monitor_cycle( + bot=bot, + config={}, + chat_ids=["chat"], + cities=["ankara"], + state=state, + focus_digest_enabled=False, + focus_digest_interval_sec=1800, + focus_digest_top_n=5, + alert_cooldown_sec=1800, + mispricing_interval_sec=7200, + min_severity="medium", + min_trigger_count=2, + mispricing_only=True, + sleep_between_cities_sec=0, + ) + + assert dirty is True + assert [row["text"] for row in bot.messages] == ["CRITICAL ankara"] + assert state["last_by_city"]["ankara"]["active"] is True + + +def test_market_monitor_cycle_excludes_critical_alert_city_from_digest(monkeypatch): + payloads = { + "critical": _focus_payload( + city="critical", + severity="high", + trigger_count=2, + signal_label="BUY YES", + yes_buy=0.05, + edge_percent=15.0, + ), + "digest a": _focus_payload(city="digest a", edge_percent=5.0, yes_buy=0.18), + "digest b": _focus_payload(city="digest b", edge_percent=4.0, yes_buy=0.2), + } + monkeypatch.setattr( + "src.utils.telegram_push.build_trade_alert_for_city", + lambda city, config: payloads[city], + ) + bot = _DigestBot() + state = {} + + dirty = _run_market_monitor_cycle( + bot=bot, + config={}, + chat_ids=["chat"], + cities=["critical", "digest a", "digest b"], + state=state, + focus_digest_enabled=True, + focus_digest_interval_sec=1800, + focus_digest_top_n=3, + alert_cooldown_sec=1800, + mispricing_interval_sec=7200, + min_severity="medium", + min_trigger_count=2, + mispricing_only=True, + sleep_between_cities_sec=0, + ) + + assert dirty is True + assert len(bot.messages) == 2 + assert bot.messages[0]["text"] == "CRITICAL critical" + digest_text = bot.messages[1]["text"] + assert "Critical" not in digest_text + assert "Digest A" in digest_text + assert "Digest B" in digest_text