diff --git a/.env.example b/.env.example index edebb1b6..c41552ac 100644 --- a/.env.example +++ b/.env.example @@ -101,14 +101,6 @@ TELEGRAM_ALERT_PUSH_INTERVAL_SEC=300 TELEGRAM_ALERT_PUSH_COOLDOWN_SEC=1800 TELEGRAM_ALERT_MIN_TRIGGER_COUNT=2 TELEGRAM_ALERT_MIN_SEVERITY=medium -TELEGRAM_ALERT_MISPRICING_ONLY=true -TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC=7200 -TELEGRAM_MARKET_FOCUS_DIGEST_ENABLED=true -TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC=1800 -TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N=5 -TELEGRAM_MARKET_FOCUS_PUSH_START_LOCAL_HOUR=8 -TELEGRAM_MARKET_FOCUS_PUSH_BEFORE_PEAK_MIN=480 -TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN=300 TELEGRAM_ALERT_CITIES=ankara,london,paris,seoul,hong kong,shanghai,singapore,tokyo,tel aviv,toronto,buenos aires,wellington,new york,chicago,dallas,miami,atlanta,seattle,lucknow,sao paulo,munich POLYWEATHER_MONITORING_ALERT_CHAT_IDS= diff --git a/src/analysis/market_alert_engine.py b/src/analysis/market_alert_engine.py index 23697653..3cb1ef9a 100644 --- a/src/analysis/market_alert_engine.py +++ b/src/analysis/market_alert_engine.py @@ -812,57 +812,6 @@ def _build_advice_cn( return "，".join(parts) + "。" -def _classify_low_yes_signal( - rules: Dict[str, Dict[str, Any]], - market_snapshot: Optional[Dict[str, Any]] = None, -) -> Dict[str, Any]: - snapshot = market_snapshot or {} - forecast_bucket = snapshot.get("forecast_bucket") or {} - yes_buy = _norm_probability(forecast_bucket.get("yes_buy")) - center_deb = rules.get("ankara_center_deb_hit", {}) or {} - momentum = rules.get("momentum_spike", {}) or {} - advection = rules.get("advection", {}) or {} - breakthrough = rules.get("forecast_breakthrough", {}) or {} - - cheap_yes = yes_buy is not None and yes_buy < 0.10 - center_hit = bool(center_deb.get("triggered")) - momentum_direction = str(momentum.get("direction") or "neutral").lower() - momentum_up = bool(momentum.get("triggered")) and momentum_direction == "up" - momentum_down = bool(momentum.get("triggered")) and momentum_direction == "down" - warm_flow = bool(advection.get("triggered")) - model_break = bool(breakthrough.get("triggered")) - - bullish_confirmation = momentum_up or warm_flow or model_break - bearish_conflict = momentum_down and not warm_flow and not model_break - - if cheap_yes and center_hit and bullish_confirmation: - return { - "label": "undervalued", - "should_push": True, - "reason_cn": "该桶 Yes 价格 < 10c，且天气信号与上行方向一致，疑似低估", - } - - if cheap_yes and center_hit and bearish_conflict: - return { - "label": "conflicted", - "should_push": False, - "reason_cn": "该桶 Yes 价格虽 < 10c，但短时动量转弱，暂不判定低估", - } - - if cheap_yes: - return { - "label": "watch", - "should_push": False, - "reason_cn": "该桶 Yes 价格 < 10c，但方向确认不足，先观察", - } - - return { - "label": "none", - "should_push": False, - "reason_cn": "当前未出现明确的低价错配信号", - } - - def _build_telegram_messages( city_weather: Dict[str, Any], rules: Dict[str, Dict[str, Any]], @@ -1057,143 +1006,6 @@ 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, - suppression: Optional[Dict[str, Any]] = None, - map_url: Optional[str] = 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", {}) - center_deb = rules.get("ankara_center_deb_hit", {}) - signal_state = _classify_low_yes_signal(rules, snapshot) - local_time = str(city_weather.get("local_time") or "").strip() - obs_time = str(current.get("obs_time") or "").strip() - suppressed = bool((suppression or {}).get("suppressed")) - has_active_trigger = any(rule.get("triggered") for rule in rules.values()) - types_cn = "高温已过（暂停推送）" if suppressed else (_join_trigger_types_cn(rules) or "天气状态快照") - - 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}" - ) - - anchor_high_c = _sf(snapshot.get("anchor_today_high_c")) - if anchor_high_c is None: - anchor_high_c = _sf(snapshot.get("open_meteo_today_high_c")) - anchor_settle = snapshot.get("anchor_settlement") - if anchor_settle is None: - anchor_settle = snapshot.get("open_meteo_settlement") - anchor_model = str(snapshot.get("anchor_model") or "").strip() - forecast_bucket = snapshot.get("forecast_bucket") or {} - match_bucket_label = str(forecast_bucket.get("label") or "--").strip() or "--" - match_bucket_yes_prob = _norm_probability(forecast_bucket.get("yes_buy")) - match_bucket_yes = ( - _fmt_cents(match_bucket_yes_prob) - if match_bucket_yes_prob is not None and match_bucket_yes_prob > 0.0 - else "--" - ) - market_url = str( - snapshot.get("market_url") - or snapshot.get("primary_market_url") - or "" - ).strip() - final_map = map_url or "https://polyweather-pro.vercel.app/" - advice = _build_advice_cn(rules, temp_symbol, suppression=suppression) - - center_line = "" - if center_deb.get("triggered"): - center_station = center_deb.get("center_station") or {} - 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_line = ( - f"Center信号：{center_name} {center_temp:.1f}{temp_symbol} 已达到 DEB {deb_prediction:.1f}{temp_symbol}" - ) - if airport_temp is not None: - center_line += f" | 机场 {airport_temp:.1f}{temp_symbol}" - if lead_gap is not None: - center_line += f" | 领先 {lead_gap:+.1f}{temp_symbol}" - - if snapshot.get("available") and signal_state.get("should_push"): - title_zh = "🚨 PolyWeather 市场监控" - elif snapshot.get("available"): - title_zh = "📍 PolyWeather 市场观察" - else: - title_zh = "🚨 PolyWeather 市场提醒" if (has_active_trigger or suppressed) else "📍 PolyWeather 状态快照" - - lines_zh = [f"{title_zh} [{city_name}]"] - lines_zh.append("") - lines_zh.append(f"类型：{types_cn}") - if anchor_high_c is not None and anchor_settle is not None: - if anchor_model: - lines_zh.append( - f"基准：多模型最高温 {anchor_model} {anchor_high_c:.1f}°C （结算参考 {anchor_settle}°C ）" - ) - else: - lines_zh.append( - f"基准：多模型最高温 {anchor_high_c:.1f}C（结算参考 {anchor_settle}C）" - ) - else: - lines_zh.append("基准：多模型最高温 --（结算参考 --）") - lines_zh.append(f"命中桶：{match_bucket_label} | Yes: {match_bucket_yes}") - lines_zh.append(f"触发：{signal_state.get('reason_cn')}") - if center_line: - lines_zh.append(center_line) - 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(f"建议：{advice}") - lines_zh.append("") - if market_url: - lines_zh.append(f"市场链接：{market_url}") - lines_zh.append(f"地图：{final_map}") - - title_en = ( - "🚨 PolyWeather Market Monitor" - if snapshot.get("available") and signal_state.get("should_push") - else "📍 PolyWeather Market Watch" - if snapshot.get("available") - else "🚨 PolyWeather Alert" - ) - lines_en = [ - f"{title_en} [{city_name}]", - "", - f"Type: {types_cn}", - f"Now: {dynamic_text}", - ] - if market_url: - lines_en.append(f"Market link: {market_url}") - 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: @@ -1234,7 +1046,6 @@ def _build_alert_evidence( trigger_types = [row.get("type") for row in triggered if row.get("type")] forecast_bucket = market_snapshot.get("forecast_bucket") or {} - low_yes_signal = _classify_low_yes_signal(rules, market_snapshot) return { "version": 1, @@ -1315,7 +1126,11 @@ def _build_alert_evidence( }, "market": { "available": bool(market_snapshot.get("available")), - "low_yes_signal": low_yes_signal, + "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"), @@ -1358,7 +1173,6 @@ def build_trading_alerts( city = city_weather.get("name", "") now = datetime.now(timezone.utc).isoformat() market_snapshot = _extract_market_snapshot(city_weather) - low_yes_signal = _classify_low_yes_signal({}, {}) rules: Dict[str, Dict[str, Any]] = { "ankara_center_deb_hit": _calc_ankara_center_deb_alert(city_weather, temp_symbol), @@ -1366,7 +1180,6 @@ def build_trading_alerts( "forecast_breakthrough": _calc_forecast_breakthrough_alert(city_weather, temp_symbol), "advection": _calc_advection_alert(city_weather, temp_symbol), } - low_yes_signal = _classify_low_yes_signal(rules, market_snapshot) triggered = [ { @@ -1396,22 +1209,13 @@ def build_trading_alerts( if force_push and severity == "none": severity = "medium" - if market_snapshot.get("available") and low_yes_signal.get("should_push"): - telegram = _build_telegram_messages_mispricing( - city_weather=city_weather, - rules=rules, - market_snapshot=market_snapshot, - suppression=suppression, - map_url=map_url, - ) - else: - telegram = _build_telegram_messages( - city_weather=city_weather, - rules=rules, - map_url=map_url, - market_snapshot=market_snapshot, - suppression=suppression, - ) + 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, diff --git a/src/bot/handlers/basic.py b/src/bot/handlers/basic.py index 4d22068f..47652a11 100644 --- a/src/bot/handlers/basic.py +++ b/src/bot/handlers/basic.py @@ -1,6 +1,5 @@ from __future__ import annotations -import threading from typing import Any from typing import Callable @@ -252,42 +251,17 @@ class BasicCommandHandler: if chat_type and chat_type != "private": self.bot.reply_to( message, - "ℹ️ `/markets` 仅支持私聊机器人查询。\n频道继续接收自动推送；如需手动查看当前市场概览，请私聊 bot 发送 `/markets`。", + "ℹ️ `/markets` 仅支持私聊机器人查询。", parse_mode="Markdown", ) trace.set_status("blocked", f"unsupported_chat_type:{chat_type}") return - chat_id = getattr(getattr(message, "chat", None), "id", None) - from src.utils.telegram_push import ( - build_market_monitor_digest, - load_cached_market_monitor_digest, + self.bot.reply_to( + message, + "ℹ️ 市场概览 (Focus Digest) 功能已移除。\n频道继续接收关键市场警报推送；如需查看当前市场状态，请访问 https://polyweather-pro.vercel.app/", + disable_web_page_preview=True, ) - - cached_summary = load_cached_market_monitor_digest() - if cached_summary: - self.bot.reply_to(message, cached_summary, disable_web_page_preview=True) - else: - self.bot.reply_to(message, "⏳ 正在生成当前市场概览，请稍候...") - - def _worker() -> None: - try: - summary = build_market_monitor_digest( - self.config, - slot_label="当前市场概览", - force_refresh=False, - ) - if not cached_summary or summary.strip() != cached_summary.strip(): - self.bot.send_message(chat_id, summary, disable_web_page_preview=True) - except Exception: - if not cached_summary: - self.bot.send_message(chat_id, "❌ 当前市场概览生成失败，请稍后重试。") - - threading.Thread( - target=_worker, - name="telegram-markets-manual-query", - daemon=True, - ).start() - trace.set_status("accepted") + trace.set_status("ok", "removed") finally: trace.emit() diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index cef1ac87..940a84dc 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -157,7 +157,7 @@ class StartupCoordinator: 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-and-focus-digest", + "mode": "critical-alerts", "interval_sec": interval, "cities_count": cities_count, "chat_targets": len(chat_ids), @@ -165,31 +165,8 @@ class StartupCoordinator: 60, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 21600), ), - "mispricing_interval_sec": max( - max(60, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 21600)), - _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 43200), - ), "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(), - "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, - _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 21600), - ), - "focus_push_start_local_hour": min( - 23, - max(0, _env_int("TELEGRAM_MARKET_FOCUS_PUSH_START_LOCAL_HOUR", 8)), - ), - "focus_push_before_peak_min": max( - 0, - _env_int("TELEGRAM_MARKET_FOCUS_PUSH_BEFORE_PEAK_MIN", 480), - ), - "focus_push_after_peak_min": max( - 0, - _env_int("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", 300), - ), - "schedule_basis": "city_local_time_for_push_window", "daily_signal_cap": "none", } validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS" diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index 3055f9b1..8c4429a4 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -234,7 +234,7 @@ def _save_state(path: str, state: Dict[str, Any]) -> None: def _cleanup_state(state: Dict[str, Any], now_ts: int, keep_sec: int = 7 * 86400) -> None: - for bucket_name in ("by_signature", "focus_digest_slots"): + for bucket_name in ("by_signature",): bucket = state.get(bucket_name, {}) if not isinstance(bucket, dict): state[bucket_name] = {} @@ -256,528 +256,6 @@ def _cleanup_state(state: Dict[str, Any], now_ts: int, keep_sec: int = 7 * 86400 last_by_city.pop(city, None) -def _cache_market_monitor_digest( - state: Dict[str, Any], - *, - message: str, - slot_label: str, - generated_at_ts: Optional[int] = None, -) -> None: - state["last_market_monitor_digest"] = { - "message": str(message or "").strip(), - "slot_label": str(slot_label or "").strip(), - "generated_at_ts": int(generated_at_ts or time.time()), - } - - -def load_cached_market_monitor_digest() -> str: - state = _load_state(_state_file()) - cached = state.get("last_market_monitor_digest") or {} - if not isinstance(cached, dict): - return "" - return str(cached.get("message") or "").strip() - - -def _minute_of_day(text: Optional[str]) -> Optional[int]: - raw = str(text or "").strip() - if not raw or ":" not in raw: - return None - try: - hour_s, minute_s = raw.split(":", 1) - hour = int(hour_s) - minute = int(minute_s[:2]) - except Exception: - return None - if not (0 <= hour <= 23 and 0 <= minute <= 59): - return None - return hour * 60 + minute - - -def _format_minutes_window(delta_minutes: int) -> str: - total = abs(int(delta_minutes)) - hours = total // 60 - minutes = total % 60 - if hours > 0 and minutes > 0: - text = f"{hours}h{minutes:02d}m" - elif hours > 0: - text = f"{hours}h" - else: - text = f"{minutes}m" - return text - - -def _format_interval_brief(seconds: int) -> str: - total = max(1, int(seconds)) - if total % 3600 == 0: - hours = total // 3600 - return f"{hours}小时" - if total % 60 == 0: - minutes = total // 60 - return f"{minutes}分钟" - return f"{total}秒" - - -def _focus_push_window_config() -> Dict[str, int]: - return { - "start_local_hour": min( - 23, - max(0, _env_int("TELEGRAM_MARKET_FOCUS_PUSH_START_LOCAL_HOUR", 8)), - ), - "before_peak_min": max( - 0, - _env_int("TELEGRAM_MARKET_FOCUS_PUSH_BEFORE_PEAK_MIN", 480), - ), - "after_peak_min": max( - 0, - _env_int("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", 300), - ), - } - - -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'])} 内" - ) - - -def _local_peak_context(alert_payload: Dict[str, Any]) -> Dict[str, Any]: - evidence = alert_payload.get("evidence") or {} - generated_local_time = str(evidence.get("generated_local_time") or "").strip() - trigger_summary = evidence.get("trigger_summary") or {} - suppression_snapshot = trigger_summary.get("suppression_snapshot") or {} - peak_time = str(suppression_snapshot.get("max_temp_time") or "").strip() - - local_min = _minute_of_day(generated_local_time) - peak_min = _minute_of_day(peak_time) - if local_min is None or peak_min is None: - return { - "local_time": generated_local_time, - "peak_time": peak_time, - "minutes_to_peak": None, - "score_adjustment": 0.0, - "window_label": "", - } - - delta = peak_min - local_min - score = 0.0 - window_label = "" - if 0 <= delta <= 120: - score = 18.0 - window_label = f"峰值前 {_format_minutes_window(delta)}" - elif 120 < delta <= 360: - score = 10.0 - window_label = f"距峰值 {_format_minutes_window(delta)}" - elif -90 <= delta < 0: - score = 6.0 - window_label = f"峰值后 {_format_minutes_window(delta)}" - elif delta < -90: - score = -8.0 - window_label = "峰值已过较久" - - return { - "local_time": generated_local_time, - "peak_time": peak_time, - "minutes_to_peak": delta, - "score_adjustment": score, - "window_label": window_label, - } - - -def _market_monitor_score(alert_payload: Dict[str, Any]) -> float: - severity = str(alert_payload.get("severity") or "none").lower() - severity_score = {"high": 36.0, "medium": 24.0, "none": 0.0}.get(severity, 0.0) - trigger_count = int(alert_payload.get("trigger_count") or 0) - trigger_score = min(18.0, float(trigger_count) * 9.0) - - snapshot = alert_payload.get("market_snapshot") or {} - if not isinstance(snapshot, dict): - snapshot = {} - if not snapshot.get("available"): - return 0.0 - - edge_percent = abs(_safe_float(snapshot.get("edge_percent")) or 0.0) - edge_score = min(22.0, edge_percent * 2.5) - - yes_buy = _norm_prob(snapshot.get("yes_buy")) - if yes_buy is None: - forecast_bucket = snapshot.get("forecast_bucket") or {} - if isinstance(forecast_bucket, dict): - yes_buy = _norm_prob(forecast_bucket.get("yes_buy")) - pricing_score = 0.0 - if yes_buy is not None: - if yes_buy < 0.10: - pricing_score = 14.0 - elif yes_buy < 0.20: - pricing_score = 9.0 - elif yes_buy < 0.35: - pricing_score = 5.0 - - confidence = str(snapshot.get("confidence") or "").strip().lower() - confidence_score = {"high": 10.0, "medium": 6.0, "low": 2.0}.get(confidence, 0.0) - - suppression = alert_payload.get("suppression") or {} - suppressed_penalty = -20.0 if bool(suppression.get("suppressed")) else 0.0 - peak_context = _local_peak_context(alert_payload) - - return max( - 0.0, - severity_score - + trigger_score - + edge_score - + pricing_score - + confidence_score - + suppressed_penalty - + float(peak_context.get("score_adjustment") or 0.0), - ) - - -def _priority_label(score: float) -> str: - if score >= 72: - return "高优先级" - if score >= 48: - return "重点观察" - return "继续观察" - - -def _join_trigger_types_cn_local(rules: Dict[str, Dict[str, Any]]) -> str: - label_map = { - "ankara_center_deb_hit": "中心站触及 DEB", - "momentum_spike": "短时动量异动", - "forecast_breakthrough": "实测击穿模型", - "advection": "暖平流信号", - } - parts: List[str] = [] - for key, label in label_map.items(): - row = rules.get(key) or {} - if row.get("triggered"): - parts.append(label) - return " + ".join(parts) - - -def _focus_trigger_summary(alert_payload: Dict[str, Any]) -> str: - rules = alert_payload.get("rules") or {} - if not isinstance(rules, dict): - return "市场与天气分歧待观察" - 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]], - *, - top_n: int, - for_push: bool = False, -) -> List[Dict[str, Any]]: - ranked = sorted( - payloads, - key=lambda item: _market_monitor_score(item), - reverse=True, - ) - shortlisted: List[Dict[str, Any]] = [] - for item in ranked: - if not bool((item.get("market_snapshot") or {}).get("available")): - continue - if not _market_price_cap_ok(item, require_actionable_quote=True): - continue - if for_push: - window_config = _focus_push_window_config() - peak_context = _local_peak_context(item) - local_min = _minute_of_day(peak_context.get("local_time")) - minutes_to_peak = peak_context.get("minutes_to_peak") - if local_min is not None and local_min < window_config["start_local_hour"] * 60: - continue - if isinstance(minutes_to_peak, (int, float)): - if minutes_to_peak > window_config["before_peak_min"]: - continue - if minutes_to_peak < -window_config["after_peak_min"]: - continue - shortlisted.append(item) - if len(shortlisted) >= top_n: - break - return shortlisted - - -def _build_focus_digest_message( - payloads: List[Dict[str, Any]], - *, - slot_label: str, - top_n: int, -) -> str: - scan_interval = _format_interval_brief(_env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 300)) - digest_interval = _format_interval_brief( - _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", 1800), - ) - shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n, for_push=True) - if not shortlisted: - return "" - - lines = [ - f"🌐 PolyWeather 市场监控 · {slot_label}", - "", - ] - - for idx, payload in enumerate(shortlisted, start=1): - city = str(payload.get("city") or "").strip().lower() - city_name = (CITY_REGISTRY.get(city) or {}).get("display_name") or city.title() or "--" - snapshot = payload.get("market_snapshot") or {} - evidence = payload.get("evidence") or {} - inputs = evidence.get("inputs") or {} - - bucket = str( - (snapshot.get("forecast_bucket") or {}).get("label") - or snapshot.get("top_bucket") - or "--" - ).strip() - current_temp = _safe_float(inputs.get("current_temp")) - deb_prediction = _safe_float(inputs.get("deb_prediction")) - market_url = str(snapshot.get("market_url") or snapshot.get("primary_market_url") or "").strip() - peak_context = _local_peak_context(payload) - - 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() - if local_time or peak_time or window_label: - context_parts: List[str] = [] - if local_time: - context_parts.append(f"当地 {local_time}") - if peak_time: - context_parts.append(f"日内高点参考 {peak_time}") - if window_label: - context_parts.append(window_label) - lines.append(" " + " | ".join(context_parts)) - if current_temp is not None or deb_prediction is not None: - lines.append( - " " - + (f"实测 {current_temp:.1f}°C" if current_temp is not None else "实测 --") - + " | " - + ( - f"DEB 预报 {deb_prediction:.1f}°C" - if deb_prediction is not None - else "DEB 预报 --" - ) - ) - lines.append(f" 触发：{_focus_trigger_summary(payload)}") - if market_url: - lines.append(f" 链接：{market_url}") - lines.append("") - - frequency_parts = [ - f"后台扫描：约每{scan_interval}一次", - f"主动推送：约每{digest_interval}一次", - f"观察窗口：{_focus_push_window_label()}", - "没有每日信号次数上限", - ] - lines.append("更新频率：" + "；".join(frequency_parts)) - return "\n".join(lines).strip() - - -def _maybe_send_focus_digest( - bot: Any, - chat_ids: List[str], - payloads: List[Dict[str, Any]], - state: Dict[str, Any], - *, - digest_interval_sec: int, - top_n: int, -) -> bool: - if not chat_ids or not payloads or digest_interval_sec <= 0: - logger.info( - "market focus digest skipped reason=invalid_runtime chat_targets={} payloads={} interval_sec={}", - len(chat_ids), - len(payloads), - digest_interval_sec, - ) - return False - - now_ts = int(time.time()) - last_digest_ts = int(state.get("last_focus_digest_ts") or 0) - available_count = sum( - 1 for item in payloads - if bool((item.get("market_snapshot") or {}).get("available")) - ) - actionable_count = sum( - 1 for item in payloads - if bool((item.get("market_snapshot") or {}).get("available")) - and _market_price_cap_ok(item, require_actionable_quote=True) - ) - shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n, for_push=True) - reference_shortlisted = _shortlist_focus_payloads(payloads, top_n=top_n, for_push=False) - high_priority_count = sum(1 for item in shortlisted if _market_monitor_score(item) >= 72) - logger.info( - "market focus digest evaluate payloads={} available={} actionable={} shortlisted={} reference_shortlisted={} high_priority={} interval_sec={} top_n={} push_window={}", - len(payloads), - available_count, - actionable_count, - len(shortlisted), - len(reference_shortlisted), - high_priority_count, - digest_interval_sec, - top_n, - _focus_push_window_label(), - ) - if last_digest_ts and now_ts - last_digest_ts < digest_interval_sec: - logger.info( - "market focus digest skipped reason=cooldown elapsed_sec={} required_sec={} shortlisted={} high_priority={}", - now_ts - last_digest_ts, - digest_interval_sec, - len(shortlisted), - high_priority_count, - ) - return False - - if not shortlisted: - logger.info( - "market focus digest skipped reason=no_candidates payloads={} available={} actionable={} top_n={}", - len(payloads), - available_count, - actionable_count, - top_n, - ) - return False - # Tighten channel pushes: require either multiple candidates or one truly high-priority market. - if len(shortlisted) < 2 and high_priority_count < 1: - first_city = str(shortlisted[0].get("city") or "--") if shortlisted else "--" - first_score = _market_monitor_score(shortlisted[0]) if shortlisted else 0 - logger.info( - "market focus digest skipped reason=too_few_candidates shortlisted={} high_priority={} first_city={} first_score={}", - len(shortlisted), - high_priority_count, - first_city, - first_score, - ) - return False - - local_now = datetime.now().astimezone() - hour = local_now.hour - if 6 <= hour < 15: - slot_label = "白天关注" - elif 15 <= hour < 23: - slot_label = "今晚关注" - else: - slot_label = "夜间关注" - message = _build_focus_digest_message( - shortlisted, - slot_label=slot_label, - top_n=top_n, - ) - if not message: - logger.info( - "market focus digest skipped reason=empty_message shortlisted={} slot_label={}", - len(shortlisted), - slot_label, - ) - return False - - _cache_market_monitor_digest( - state, - message=message, - slot_label=slot_label, - ) - - sent_count = 0 - for chat_id in chat_ids: - try: - bot.send_message(chat_id, message, disable_web_page_preview=True) - sent_count += 1 - except Exception as exc: - logger.warning( - "market focus digest push failed interval_sec={} chat_id={} error={}", - digest_interval_sec, - chat_id, - exc, - ) - if sent_count <= 0: - return False - - state["last_focus_digest_ts"] = now_ts - logger.info( - "market focus digest pushed interval_sec={} shortlisted={} payloads={} chat_targets={} slot_label={}", - digest_interval_sec, - len(shortlisted), - len(payloads), - sent_count, - slot_label, - ) - return True - - -def build_market_monitor_digest( - config: Dict[str, Any], - *, - slot_label: str = "当前概览", - top_n: Optional[int] = None, - force_refresh: bool = False, -) -> str: - cities = _parse_city_list(os.getenv("TELEGRAM_ALERT_CITIES")) - - digest_top_n = top_n if top_n is not None else max( - 3, - min(8, _env_int("TELEGRAM_MARKET_FOCUS_DIGEST_TOP_N", 5)), - ) - payloads: List[Dict[str, Any]] = [] - for city in cities: - try: - payloads.append(build_trade_alert_for_city(city, config, force_refresh=force_refresh)) - except Exception as exc: - logger.warning("market monitor digest build skipped city={} error={}", city, exc) - message = _build_focus_digest_message( - payloads, - slot_label=slot_label, - top_n=digest_top_n, - ) - if message: - state = _load_state(_state_file()) - _cache_market_monitor_digest( - state, - message=message, - slot_label=slot_label, - ) - _save_state(_state_file(), state) - return message - return "ℹ️ 当前没有可用的市场监控摘要。" - - def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_count: int) -> bool: triggered_alerts = alert_payload.get("triggered_alerts") or [] if any(alert.get("force_push") for alert in triggered_alerts): @@ -792,17 +270,9 @@ def _severity_ok(alert_payload: Dict[str, Any], min_severity: str, min_trigger_c def _market_price_cap_ok( alert_payload: Dict[str, Any], - require_actionable_quote: bool = False, ) -> bool: market = alert_payload.get("market_snapshot") or {} if not isinstance(market, dict) or not market.get("available"): - if require_actionable_quote: - logger.info( - "trade alert skipped: market snapshot unavailable city={}".format( - alert_payload.get("city"), - ) - ) - return False return True primary_market = market.get("primary_market") or {} @@ -1057,18 +527,12 @@ def _maybe_send_alert( cooldown_sec: int, min_severity: str, min_trigger_count: int, - mispricing_only: bool, ) -> 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 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, - ): + if not _market_price_cap_ok(alert_payload): is_active = False message = ((alert_payload.get("telegram") or {}).get("zh") or "").strip() @@ -1129,27 +593,6 @@ 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], @@ -1157,64 +600,32 @@ def _run_market_monitor_cycle( 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, + cooldown_sec=alert_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 @@ -1232,33 +643,20 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th 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)) - mispricing_interval_sec = max( - alert_cooldown_sec, - _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 43200), - ) 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" - 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", 21600), - ) 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-and-focus-digest " + 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 mispricing_interval={mispricing_interval_sec}s " + f"alert_cooldown={alert_cooldown_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"state_path={state_path}" ) while True: @@ -1271,14 +669,9 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th 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) diff --git a/tests/test_market_alert_engine.py b/tests/test_market_alert_engine.py index a01202d3..8d785a49 100644 --- a/tests/test_market_alert_engine.py +++ b/tests/test_market_alert_engine.py @@ -1,10 +1,5 @@ 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, -) +from src.utils.telegram_push import _run_market_monitor_cycle def _sample_weather_payload(): @@ -181,27 +176,6 @@ def test_peak_passed_guard_suppresses_late_day_cooldown_alerts(): assert "暂停主动推送" in out["telegram"]["zh"] -def test_market_monitor_digest_skips_non_tradable_market(monkeypatch): - payload = build_trading_alerts( - city_weather=_sample_weather_payload(), - map_url="https://example.com/map", - ) - payload["market_snapshot"]["available"] = True - payload["market_snapshot"]["market_closed"] = True - payload["market_snapshot"]["market_tradable"] = False - payload["market_snapshot"]["market_accepting_orders"] = False - payload["market_snapshot"]["market_tradable_reason"] = "in_review" - - monkeypatch.setattr( - "src.utils.telegram_push.build_trade_alert_for_city", - lambda city, config, force_refresh=False: payload, - ) - monkeypatch.setenv("TELEGRAM_ALERT_CITIES", "ankara") - - digest = build_market_monitor_digest({}, slot_label="当前市场概览") - assert digest == "ℹ️ 当前没有可用的市场监控摘要。" - - class _DigestBot: def __init__(self): self.messages = [] @@ -285,95 +259,6 @@ def _focus_payload( } -def test_focus_digest_push_window_does_not_cut_off_after_2pm_spanish_time(monkeypatch): - monkeypatch.delenv("TELEGRAM_MARKET_FOCUS_PUSH_AFTER_PEAK_MIN", raising=False) - - shortlisted = _shortlist_focus_payloads( - [_focus_payload(local_time="14:00", peak_time="12:00")], - top_n=5, - for_push=True, - ) - - assert len(shortlisted) == 1 - - -def test_focus_digest_footer_explains_no_daily_signal_cap(monkeypatch): - monkeypatch.setenv("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", "300") - monkeypatch.setenv("TELEGRAM_MARKET_FOCUS_DIGEST_INTERVAL_SEC", "1800") - - message = _build_focus_digest_message( - [_focus_payload(local_time="14:00", peak_time="12:00")], - slot_label="白天关注", - top_n=5, - ) - - 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", @@ -396,14 +281,9 @@ def test_market_monitor_cycle_restores_critical_alert_push(monkeypatch): 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, ) @@ -412,47 +292,3 @@ def test_market_monitor_cycle_restores_critical_alert_push(monkeypatch): 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