gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

移除错误定价信号和市场聚焦摘要功能

Browse Source 
保留关键市场警报推送（4 条触发规则不变），
移除 mispricing 信号分类/推送、focus digest 评分/聚合/定时推送、
/markets 手动查询等全套逻辑，净删除 1049 行。
This commit is contained in:
2569718930@qq.com
2026-05-07 21:42:06 +08:00
parent 30faac989c
commit 2183c0399d
6 changed files with 25 additions and 1049 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
-8
View File
@@ -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=
src/analysis/market_alert_engine.py
+12 -208
View File
@@ -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,
src/bot/handlers/basic.py
+6 -32
View File
@@ -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()
src/bot/runtime_coordinator.py
+1 -24
View File
@@ -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"
src/utils/telegram_push.py
+5 -612
View File
@@ -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)
tests/test_market_alert_engine.py
+1 -165
View File
@@ -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