移除 PolyWeather 市场提醒功能及监控基础设施
- 删除 market_alert_engine.py：交易预警引擎 - 删除 alertmanager_telegram_relay.py：Alertmanager 到 Telegram 转发 - 移除 telegram_push.py 中市场监控推送循环 - 移除 runtime_coordinator.py 中 trade_alert_push 协程 - 移除 docker-compose.yml 中 Prometheus/Alertmanager/Grafana 服务 - 移除 monitoring/ 目录：prometheus/alertmanager/grafana 配置 Tested: ruff check . ✓
This commit is contained in:
2569718930@qq.com
@@ -483,211 +483,6 @@ def _alert_signature(alert_payload: Dict[str, Any]) -> str:
|
||||
raw = json.dumps(signature_payload, sort_keys=True, ensure_ascii=True)
|
||||
return hashlib.sha1(raw.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def build_trade_alert_for_city(
|
||||
city: str,
|
||||
config: Dict[str, Any],
|
||||
force_refresh: bool = False,
|
||||
target_date: Optional[str] = None,
|
||||
) -> Dict[str, Any]:
|
||||
from web.app import _analyze, _build_city_detail_payload
|
||||
from src.analysis.market_alert_engine import build_trading_alerts
|
||||
|
||||
city_weather = _analyze(city, force_refresh=force_refresh)
|
||||
try:
|
||||
aggregate_detail = _build_city_detail_payload(
|
||||
city_weather,
|
||||
target_date=target_date,
|
||||
)
|
||||
market_scan = aggregate_detail.get("market_scan")
|
||||
if isinstance(market_scan, dict):
|
||||
city_weather = {**city_weather, "market_scan": market_scan}
|
||||
except Exception as exc:
|
||||
logger.debug(f"market scan attach skipped city={city}: {exc}")
|
||||
|
||||
resolved_target_date = target_date or city_weather.get("local_date")
|
||||
if resolved_target_date:
|
||||
datetime.strptime(resolved_target_date, "%Y-%m-%d")
|
||||
|
||||
map_url = os.getenv("POLYWEATHER_MAP_URL") or "https://polyweather-pro.vercel.app/"
|
||||
alert_payload = build_trading_alerts(
|
||||
city_weather=city_weather,
|
||||
map_url=map_url,
|
||||
)
|
||||
alert_payload["target_date"] = resolved_target_date
|
||||
return alert_payload
|
||||
|
||||
|
||||
def _maybe_send_alert(
|
||||
bot: Any,
|
||||
chat_ids: List[str],
|
||||
city: str,
|
||||
alert_payload: Dict[str, Any],
|
||||
state: Dict[str, Any],
|
||||
cooldown_sec: int,
|
||||
min_severity: str,
|
||||
min_trigger_count: int,
|
||||
) -> bool:
|
||||
now_ts = int(time.time())
|
||||
last_by_city = state.setdefault("last_by_city", {})
|
||||
last_city = last_by_city.get(city) or {}
|
||||
is_active = _severity_ok(alert_payload, min_severity, min_trigger_count)
|
||||
if not _market_price_cap_ok(alert_payload):
|
||||
is_active = False
|
||||
message = ((alert_payload.get("telegram") or {}).get("zh") or "").strip()
|
||||
|
||||
if not is_active or not message:
|
||||
if last_city.get("active"):
|
||||
last_by_city[city] = {
|
||||
**last_city,
|
||||
"active": False,
|
||||
"cleared_ts": now_ts,
|
||||
}
|
||||
logger.info(f"market monitor disarmed city={city}")
|
||||
return True
|
||||
return False
|
||||
|
||||
if not chat_ids:
|
||||
return False
|
||||
|
||||
signature = _alert_signature(alert_payload)
|
||||
trigger_key = _trigger_type_key(alert_payload)
|
||||
last_city_sig = last_city.get("signature")
|
||||
last_city_key = str(last_city.get("trigger_key") or "")
|
||||
last_city_ts = int(last_city.get("ts") or 0)
|
||||
last_sig_ts = int((state.get("by_signature") or {}).get(signature) or 0)
|
||||
last_city_active = bool(last_city.get("active"))
|
||||
|
||||
if last_city_active and last_city_key == trigger_key and last_city_sig == signature:
|
||||
return False
|
||||
|
||||
if last_city_ts and now_ts - last_city_ts < cooldown_sec:
|
||||
return False
|
||||
if last_sig_ts and now_ts - last_sig_ts < cooldown_sec:
|
||||
return False
|
||||
|
||||
sent_count = 0
|
||||
for chat_id in chat_ids:
|
||||
try:
|
||||
bot.send_message(chat_id, message)
|
||||
sent_count += 1
|
||||
except Exception as exc:
|
||||
logger.warning("market monitor push failed city={} chat_id={} error={}", city, chat_id, exc)
|
||||
if sent_count <= 0:
|
||||
return False
|
||||
|
||||
last_by_city[city] = {
|
||||
"signature": signature,
|
||||
"trigger_key": trigger_key,
|
||||
"severity": alert_payload.get("severity"),
|
||||
"ts": now_ts,
|
||||
"active": True,
|
||||
"evidence": alert_payload.get("evidence"),
|
||||
}
|
||||
state.setdefault("by_signature", {})[signature] = now_ts
|
||||
logger.info(
|
||||
f"market monitor pushed city={city} severity={alert_payload.get('severity')} "
|
||||
f"trigger_count={alert_payload.get('trigger_count')} trigger_key={trigger_key} "
|
||||
f"evidence={_evidence_brief(alert_payload)} chat_targets={sent_count}"
|
||||
)
|
||||
return True
|
||||
|
||||
|
||||
def _run_market_monitor_cycle(
|
||||
bot: Any,
|
||||
config: Dict[str, Any],
|
||||
*,
|
||||
chat_ids: List[str],
|
||||
cities: List[str],
|
||||
state: Dict[str, Any],
|
||||
alert_cooldown_sec: int,
|
||||
min_severity: str,
|
||||
min_trigger_count: int,
|
||||
sleep_between_cities_sec: float = 1.0,
|
||||
) -> bool:
|
||||
state_dirty = False
|
||||
|
||||
for city in cities:
|
||||
try:
|
||||
alert_payload = build_trade_alert_for_city(city, config)
|
||||
if _maybe_send_alert(
|
||||
bot=bot,
|
||||
chat_ids=chat_ids,
|
||||
city=city,
|
||||
alert_payload=alert_payload,
|
||||
state=state,
|
||||
cooldown_sec=alert_cooldown_sec,
|
||||
min_severity=min_severity,
|
||||
min_trigger_count=min_trigger_count,
|
||||
):
|
||||
state_dirty = True
|
||||
except Exception:
|
||||
logger.exception(f"telegram market monitor loop failed for city={city}")
|
||||
if sleep_between_cities_sec > 0:
|
||||
time.sleep(sleep_between_cities_sec)
|
||||
|
||||
return state_dirty
|
||||
|
||||
|
||||
def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[threading.Thread]:
|
||||
enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True)
|
||||
chat_ids = get_telegram_chat_ids_from_env()
|
||||
if not enabled:
|
||||
logger.info("telegram market monitor loop disabled")
|
||||
return None
|
||||
if not chat_ids:
|
||||
logger.warning("telegram market monitor loop skipped: TELEGRAM_CHAT_IDS is not set")
|
||||
return None
|
||||
|
||||
interval_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_INTERVAL_SEC", 1800))
|
||||
cities = _parse_city_list(os.getenv("TELEGRAM_ALERT_CITIES"))
|
||||
state_path = _state_file()
|
||||
alert_cooldown_sec = max(60, _env_int("TELEGRAM_ALERT_PUSH_COOLDOWN_SEC", 21600))
|
||||
min_trigger_count = max(1, _env_int("TELEGRAM_ALERT_MIN_TRIGGER_COUNT", 3))
|
||||
min_severity = str(os.getenv("TELEGRAM_ALERT_MIN_SEVERITY") or "high").strip().lower()
|
||||
if min_severity not in SEVERITY_RANK:
|
||||
min_severity = "high"
|
||||
def _runner() -> None:
|
||||
try:
|
||||
_save_state(state_path, _load_state(state_path))
|
||||
except Exception:
|
||||
logger.exception(f"failed to initialize market monitor state path={state_path}")
|
||||
logger.info(
|
||||
f"telegram market monitor loop started mode=critical-alerts "
|
||||
f"cities={len(cities)} interval={interval_sec}s chat_targets={len(chat_ids)} "
|
||||
f"alert_cooldown={alert_cooldown_sec}s "
|
||||
f"min_severity={min_severity} min_trigger_count={min_trigger_count} "
|
||||
f"state_path={state_path}"
|
||||
)
|
||||
while True:
|
||||
cycle_started = time.time()
|
||||
state = _load_state(state_path)
|
||||
_cleanup_state(state, int(cycle_started))
|
||||
if _run_market_monitor_cycle(
|
||||
bot=bot,
|
||||
config=config,
|
||||
chat_ids=chat_ids,
|
||||
cities=cities,
|
||||
state=state,
|
||||
alert_cooldown_sec=alert_cooldown_sec,
|
||||
min_severity=min_severity,
|
||||
min_trigger_count=min_trigger_count,
|
||||
):
|
||||
_save_state(state_path, state)
|
||||
|
||||
elapsed = time.time() - cycle_started
|
||||
sleep_sec = max(5, interval_sec - int(elapsed))
|
||||
time.sleep(sleep_sec)
|
||||
|
||||
thread = threading.Thread(
|
||||
target=_runner,
|
||||
name="telegram-market-monitor-pusher",
|
||||
daemon=True,
|
||||
)
|
||||
thread.start()
|
||||
return thread
|
||||
|
||||
|
||||
# ── high-freq airport push loop ──
|
||||
|
||||
HIGH_FREQ_AIRPORT_CITIES = {"seoul", "busan", "tokyo", "ankara", "helsinki", "amsterdam", "istanbul", "paris", "hong kong", "lau fau shan", "taipei"}
|
||||
|
||||
Reference in New Issue
Block a user