From cd0a5d519f2117935bf9009793d3d86cd03b79bd Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Fri, 13 Mar 2026 19:24:05 +0800 Subject: [PATCH] feat: Introduce core bot components, on-chain wallet watchers, Telegram notifications, and configuration management. --- .env.example | 5 +- src/bot/command_guard.py | 58 +++++++++++-------- src/bot/orchestrator.py | 3 +- src/bot/runtime_coordinator.py | 28 +++++---- src/bot/weekly_reward_loop.py | 32 ++++++---- src/onchain/polygon_wallet_watcher.py | 32 ++++++++-- .../polymarket_wallet_activity_watcher.py | 34 ++++++++--- src/payments/contract_checkout.py | 11 +++- src/utils/config_loader.py | 8 ++- src/utils/telegram_chat_ids.py | 45 ++++++++++++++ src/utils/telegram_push.py | 34 ++++++++--- 11 files changed, 214 insertions(+), 76 deletions(-) create mode 100644 src/utils/telegram_chat_ids.py diff --git a/.env.example b/.env.example index 8f30e177..431438a8 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,9 @@ # Telegram Bot TELEGRAM_BOT_TOKEN=your_bot_token_here TELEGRAM_CHAT_ID=your_chat_id_here +# Optional multi-chat target (comma-separated). If set, it will be merged with TELEGRAM_CHAT_ID. +# Example: TELEGRAM_CHAT_IDS=-1003586303099,-1003539418691 +TELEGRAM_CHAT_IDS= TELEGRAM_ALERT_PUSH_ENABLED=true TELEGRAM_ALERT_PUSH_INTERVAL_SEC=300 TELEGRAM_ALERT_PUSH_COOLDOWN_SEC=1800 @@ -50,7 +53,7 @@ NEXT_PUBLIC_WALLETCONNECT_PROJECT_ID= NEXT_PUBLIC_WALLETCONNECT_POLYGON_RPC_URL=https://polygon-bor-rpc.publicnode.com # Bot command access guard (/city, /deb): # - Pro entitlement removed -# - user only needs to be a member of TELEGRAM_CHAT_ID group +# - user only needs to be a member of any configured group (TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID) POLYWEATHER_BOT_GROUP_INVITE_URL= # Group message points (anti-spam + ranking) POLYWEATHER_BOT_MESSAGE_POINTS=4 diff --git a/src/bot/command_guard.py b/src/bot/command_guard.py index f22cd8dc..94ebbce0 100644 --- a/src/bot/command_guard.py +++ b/src/bot/command_guard.py @@ -6,6 +6,7 @@ from typing import Any from loguru import logger from src.bot.io_layer import BotIOLayer +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env, parse_telegram_chat_ids class CommandGuard: @@ -15,7 +16,10 @@ class CommandGuard: def __init__(self, io_layer: BotIOLayer, group_chat_id: str | None = None): self.io_layer = io_layer - self.group_chat_id = str(group_chat_id or os.getenv("TELEGRAM_CHAT_ID") or "").strip() + if group_chat_id is None: + self.group_chat_ids = get_telegram_chat_ids_from_env() + else: + self.group_chat_ids = parse_telegram_chat_ids(group_chat_id) self.group_invite_url = str(os.getenv("POLYWEATHER_BOT_GROUP_INVITE_URL") or "").strip() def _reply_group_required(self, message: Any) -> None: @@ -30,42 +34,48 @@ class CommandGuard: if user_id is None: return False - if not self.group_chat_id: + if not self.group_chat_ids: logger.error( - "group member blocked command={} user_id={} reason=missing_TELEGRAM_CHAT_ID", + "group member blocked command={} user_id={} reason=missing_TELEGRAM_CHAT_IDS", command_label, user_id, ) self.io_layer.bot.reply_to( message, - "⚠️ 机器人未配置群组准入（TELEGRAM_CHAT_ID），请联系管理员。", + "⚠️ 机器人未配置群组准入（TELEGRAM_CHAT_IDS / TELEGRAM_CHAT_ID），请联系管理员。", ) return False - try: - member = self.io_layer.bot.get_chat_member(self.group_chat_id, int(user_id)) - status = str(getattr(member, "status", "") or "").strip().lower() - except Exception as exc: + blocked_statuses: list[str] = [] + lookup_failures: list[str] = [] + for chat_id in self.group_chat_ids: + try: + member = self.io_layer.bot.get_chat_member(chat_id, int(user_id)) + status = str(getattr(member, "status", "") or "").strip().lower() + except Exception as exc: + lookup_failures.append(f"{chat_id}:{type(exc).__name__}") + continue + + if status in self._ALLOWED_MEMBER_STATUSES: + return True + blocked_statuses.append(f"{chat_id}:{status or 'unknown'}") + + if blocked_statuses: logger.info( - "group member blocked command={} user_id={} chat_id={} reason=lookup_failed error={}", + "group member blocked command={} user_id={} chat_ids={} statuses={}", command_label, user_id, - self.group_chat_id, - exc, + ",".join(self.group_chat_ids), + "|".join(blocked_statuses), + ) + else: + logger.info( + "group member blocked command={} user_id={} chat_ids={} reason=lookup_failed_all errors={}", + command_label, + user_id, + ",".join(self.group_chat_ids), + "|".join(lookup_failures) or "unknown", ) - self._reply_group_required(message) - return False - - if status in self._ALLOWED_MEMBER_STATUSES: - return True - - logger.info( - "group member blocked command={} user_id={} chat_id={} status={}", - command_label, - user_id, - self.group_chat_id, - status or "unknown", - ) self._reply_group_required(message) return False diff --git a/src/bot/orchestrator.py b/src/bot/orchestrator.py index 145ab22e..44866553 100644 --- a/src/bot/orchestrator.py +++ b/src/bot/orchestrator.py @@ -16,6 +16,7 @@ from src.bot.io_layer import BotIOLayer from src.bot.runtime_coordinator import StartupCoordinator from src.bot.services.city_command_service import CityCommandService from src.bot.services.deb_command_service import DebCommandService +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env def _project_root() -> str: @@ -68,7 +69,7 @@ def start_bot() -> None: config=config, command_access_mode="group_member_only", protected_commands=["/city", "/deb"], - required_group_chat_id=str(os.getenv("TELEGRAM_CHAT_ID") or "").strip(), + required_group_chat_id=",".join(get_telegram_chat_ids_from_env()), ) _register_handlers( diff --git a/src/bot/runtime_coordinator.py b/src/bot/runtime_coordinator.py index 8bc9c54a..23bb4eb7 100644 --- a/src/bot/runtime_coordinator.py +++ b/src/bot/runtime_coordinator.py @@ -6,6 +6,8 @@ from datetime import datetime from importlib import import_module from typing import Any, Callable, Dict, List, Optional +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env + def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) @@ -148,7 +150,7 @@ class StartupCoordinator: def _start_trade_alert_loop(self) -> LoopStatus: enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True) - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() + chat_ids = get_telegram_chat_ids_from_env() mispricing_only = _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True) interval = ( max(300, _env_int("TELEGRAM_ALERT_MISPRICING_INTERVAL_SEC", 7200)) @@ -160,8 +162,9 @@ class StartupCoordinator: "mode": "mispricing-only" if mispricing_only else "full", "interval_sec": interval, "cities_count": cities_count, + "chat_targets": len(chat_ids), } - validation_error = None if chat_id else "missing_TELEGRAM_CHAT_ID" + validation_error = None if chat_ids else "missing_TELEGRAM_CHAT_IDS" return self._start_with_validation( key="trade_alert_push", label="错价雷达推送", @@ -176,7 +179,7 @@ class StartupCoordinator: def _start_polygon_wallet_loop(self) -> LoopStatus: enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False) - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() + chat_ids = get_telegram_chat_ids_from_env() rpc_url = str(os.getenv("POLYGON_RPC_URL") or "").strip() wallets_count = _parse_csv_count(os.getenv("POLYGON_WALLET_WATCH_ADDRESSES")) poll = max(3, _env_int("POLYGON_WALLET_WATCH_INTERVAL_SEC", 8)) @@ -184,10 +187,11 @@ class StartupCoordinator: "poll_sec": poll, "wallets_count": wallets_count, "polymarket_only": _env_bool("POLYGON_WALLET_WATCH_POLYMARKET_ONLY", True), + "chat_targets": len(chat_ids), } validation_error = None - if not chat_id: - validation_error = "missing_TELEGRAM_CHAT_ID" + if not chat_ids: + validation_error = "missing_TELEGRAM_CHAT_IDS" elif not rpc_url: validation_error = "missing_POLYGON_RPC_URL" elif wallets_count == 0: @@ -205,17 +209,18 @@ class StartupCoordinator: def _start_polymarket_wallet_activity_loop(self) -> LoopStatus: enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False) - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() + chat_ids = get_telegram_chat_ids_from_env() users_count = _parse_csv_count(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS")) poll = max(5, _env_int("POLYMARKET_WALLET_ACTIVITY_INTERVAL_SEC", 20)) details = { "poll_sec": poll, "users_count": users_count, "link_preview": _env_bool("POLYMARKET_WALLET_ACTIVITY_LINK_PREVIEW", True), + "chat_targets": len(chat_ids), } validation_error = None - if not chat_id: - validation_error = "missing_TELEGRAM_CHAT_ID" + if not chat_ids: + validation_error = "missing_TELEGRAM_CHAT_IDS" elif users_count == 0: validation_error = "missing_POLYMARKET_WALLET_ACTIVITY_USERS" return self._start_with_validation( @@ -231,7 +236,7 @@ class StartupCoordinator: def _start_weekly_reward_loop(self) -> LoopStatus: enabled = _env_bool("POLYWEATHER_WEEKLY_REWARD_ENABLED", True) - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() + chat_ids = get_telegram_chat_ids_from_env() settle_weekday = min( 7, max(1, _env_int("POLYWEATHER_WEEKLY_REWARD_SETTLE_WEEKDAY", 1)) ) @@ -250,11 +255,12 @@ class StartupCoordinator: "settle_time": f"{settle_hour:02d}:{settle_minute:02d}", "check_interval_sec": check_interval, "announce": _env_bool("POLYWEATHER_WEEKLY_REWARD_ANNOUNCE_ENABLED", True), + "chat_targets": len(chat_ids), } announce_enabled = bool(details["announce"]) validation_error = None - if announce_enabled and not chat_id: - validation_error = "missing_TELEGRAM_CHAT_ID" + if announce_enabled and not chat_ids: + validation_error = "missing_TELEGRAM_CHAT_IDS" return self._start_with_validation( key="weekly_reward", label="周榜奖励结算", diff --git a/src/bot/weekly_reward_loop.py b/src/bot/weekly_reward_loop.py index 11879927..c2174960 100644 --- a/src/bot/weekly_reward_loop.py +++ b/src/bot/weekly_reward_loop.py @@ -15,6 +15,7 @@ import requests from loguru import logger from src.database.db_manager import DBManager +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env def _env_bool(name: str, default: bool) -> bool: @@ -213,19 +214,20 @@ def _runner(bot: Any) -> None: interval_sec = _env_int("POLYWEATHER_WEEKLY_REWARD_CHECK_INTERVAL_SEC", 300, 30) timeout_sec = _env_int("POLYWEATHER_WEEKLY_REWARD_HTTP_TIMEOUT_SEC", 10, 3) announce = _env_bool("POLYWEATHER_WEEKLY_REWARD_ANNOUNCE_ENABLED", True) - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() + chat_ids = get_telegram_chat_ids_from_env() supabase_url = str(os.getenv("SUPABASE_URL") or "").strip().rstrip("/") service_role_key = str(os.getenv("SUPABASE_SERVICE_ROLE_KEY") or "").strip() db = DBManager() logger.info( - "weekly reward loop started tz={} settle={} {:02d}:{:02d} interval={}s announce={}", + "weekly reward loop started tz={} settle={} {:02d}:{:02d} interval={}s announce={} chat_targets={}", tz_name, settle_weekday, settle_hour, settle_minute, interval_sec, announce, + len(chat_ids), ) while True: @@ -313,16 +315,22 @@ def _runner(bot: Any) -> None: len(winners), skipped, ) - if announce and chat_id: - try: - bot.send_message( - chat_id, - _render_settle_report(week_key=week_key, winners=winners, skipped=skipped), - parse_mode="HTML", - disable_web_page_preview=True, - ) - except Exception as exc: - logger.warning(f"weekly reward announcement failed: {exc}") + if announce and chat_ids: + message = _render_settle_report(week_key=week_key, winners=winners, skipped=skipped) + for chat_id in chat_ids: + try: + bot.send_message( + chat_id, + message, + parse_mode="HTML", + disable_web_page_preview=True, + ) + except Exception as exc: + logger.warning( + "weekly reward announcement failed chat_id={} error={}", + chat_id, + exc, + ) except Exception as exc: logger.warning(f"weekly reward cycle failed: {exc}") time.sleep(interval_sec) diff --git a/src/onchain/polygon_wallet_watcher.py b/src/onchain/polygon_wallet_watcher.py index 2fcbf82d..c948225e 100644 --- a/src/onchain/polygon_wallet_watcher.py +++ b/src/onchain/polygon_wallet_watcher.py @@ -9,6 +9,8 @@ from typing import Any, Dict, List, Optional, Set, Tuple from loguru import logger from web3 import Web3 +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env + TRANSFER_TOPIC = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" APPROVAL_TOPIC = "0x8c5be1e5ebec7d5bd14f71427d1e84f3dd0314c0f7b2291e5b200ac8c7c3b925" ERC20_ABI = [ @@ -372,7 +374,7 @@ def _build_message( def start_polygon_wallet_watch_loop(bot: Any) -> Optional[threading.Thread]: enabled = _env_bool("POLYGON_WALLET_WATCH_ENABLED", False) - chat_id = os.getenv("TELEGRAM_CHAT_ID") + chat_ids = get_telegram_chat_ids_from_env() rpc_url = os.getenv("POLYGON_RPC_URL") watch_set = _parse_addresses(os.getenv("POLYGON_WALLET_WATCH_ADDRESSES")) polymarket_only = _env_bool("POLYGON_WALLET_WATCH_POLYMARKET_ONLY", True) @@ -381,8 +383,8 @@ def start_polygon_wallet_watch_loop(bot: Any) -> Optional[threading.Thread]: if not enabled: logger.info("polygon wallet watcher disabled") return None - if not chat_id: - logger.warning("polygon wallet watcher skipped: TELEGRAM_CHAT_ID is not set") + if not chat_ids: + logger.warning("polygon wallet watcher skipped: TELEGRAM_CHAT_IDS is not set") return None if not rpc_url: logger.warning("polygon wallet watcher skipped: POLYGON_RPC_URL is not set") @@ -426,7 +428,8 @@ def start_polygon_wallet_watch_loop(bot: Any) -> Optional[threading.Thread]: logger.info( f"polygon wallet watcher started wallets={len(watch_set)} " f"polymarket_only={polymarket_only} pm_contracts={len(pm_contracts)} " - f"poll={poll_sec}s confirmations={confirmations} state_path={state_path}" + f"poll={poll_sec}s confirmations={confirmations} chat_targets={len(chat_ids)} " + f"state_path={state_path}" ) while True: @@ -503,11 +506,28 @@ def start_polygon_wallet_watch_loop(bot: Any) -> Optional[threading.Thread]: approval_lines=approval_lines, tx_to_label=tx_to_label, ) - bot.send_message(chat_id, message, disable_web_page_preview=True) + 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( + "polygon wallet alert push failed wallet={} chat_id={} error={}", + matched_wallet, + chat_id, + exc, + ) + if sent_count <= 0: + continue state.setdefault("seen_tx", {})[tx_hash] = cycle_ts logger.info( f"polygon wallet alert pushed wallet={matched_wallet} " - f"tx={tx_hash} block={block_num} polymarket={pm_hit}" + f"tx={tx_hash} block={block_num} polymarket={pm_hit} chat_targets={sent_count}" ) state["last_scanned_block"] = block_num diff --git a/src/onchain/polymarket_wallet_activity_watcher.py b/src/onchain/polymarket_wallet_activity_watcher.py index 5773c66f..149a5ec0 100644 --- a/src/onchain/polymarket_wallet_activity_watcher.py +++ b/src/onchain/polymarket_wallet_activity_watcher.py @@ -9,6 +9,8 @@ from typing import Any, Dict, List, Optional, Tuple import requests from loguru import logger +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env + def _env_bool(name: str, default: bool) -> bool: raw = os.getenv(name) @@ -547,7 +549,7 @@ def _filter_changes_by_position_value( def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread]: enabled = _env_bool("POLYMARKET_WALLET_ACTIVITY_ENABLED", False) - chat_id = os.getenv("TELEGRAM_CHAT_ID") + chat_ids = get_telegram_chat_ids_from_env() users = _parse_addresses(os.getenv("POLYMARKET_WALLET_ACTIVITY_USERS")) user_aliases = _parse_address_aliases( os.getenv("POLYMARKET_WALLET_ACTIVITY_USER_ALIASES") @@ -560,8 +562,8 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread if not enabled: logger.info("polymarket wallet activity watcher disabled") return None - if not chat_id: - logger.warning("polymarket wallet activity watcher skipped: TELEGRAM_CHAT_ID is not set") + if not chat_ids: + logger.warning("polymarket wallet activity watcher skipped: TELEGRAM_CHAT_IDS is not set") return None if not users: logger.warning("polymarket wallet activity watcher skipped: POLYMARKET_WALLET_ACTIVITY_USERS is empty") @@ -627,6 +629,7 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread f"poll={poll_sec}s data_api={data_api_url} price_filter={min_price}-{max_price} " f"min_position_value_usd={min_position_value_usd} " f"min_value_exempt_users={len(exempt_wallets)} " + f"chat_targets={len(chat_ids)} " f"aliases={len(user_aliases)} link_preview={link_preview} " f"min_avg_price_delta={min_avg_price_delta} " f"immediate_on_size_delta={immediate_on_size_delta} " @@ -770,13 +773,26 @@ def start_polymarket_wallet_activity_loop(bot: Any) -> Optional[threading.Thread max_changes=max_changes, wallet_alias=user_aliases.get(user), ) - bot.send_message( - chat_id, - msg, - disable_web_page_preview=not link_preview, - ) + sent_count = 0 + for chat_id in chat_ids: + try: + bot.send_message( + chat_id, + msg, + disable_web_page_preview=not link_preview, + ) + sent_count += 1 + except Exception as exc: + logger.warning( + "wallet activity push failed user={} chat_id={} error={}", + user, + chat_id, + exc, + ) + if sent_count <= 0: + continue logger.info( - f"wallet activity pushed user={user} changes={len(outgoing)}" + f"wallet activity pushed user={user} changes={len(outgoing)} chat_targets={sent_count}" ) users_state[user] = { diff --git a/src/payments/contract_checkout.py b/src/payments/contract_checkout.py index 32cbf303..9037d5cd 100644 --- a/src/payments/contract_checkout.py +++ b/src/payments/contract_checkout.py @@ -1508,8 +1508,13 @@ class PaymentContractCheckoutService: if not self.notify_telegram: return token = str(os.getenv("TELEGRAM_BOT_TOKEN") or "").strip() - chat_id = str(os.getenv("TELEGRAM_CHAT_ID") or "").strip() - if not token or not chat_id: + if not token: + return + user = self._db.get_user_by_supabase_user_id(user_id) + if not isinstance(user, dict): + return + telegram_id = int(user.get("telegram_id") or 0) + if telegram_id <= 0: return short_hash = tx_hash[:10] + "..." + tx_hash[-8:] if len(tx_hash) > 20 else tx_hash text = ( @@ -1523,7 +1528,7 @@ class PaymentContractCheckoutService: requests.post( f"https://api.telegram.org/bot{token}/sendMessage", json={ - "chat_id": chat_id, + "chat_id": str(telegram_id), "text": text, "disable_web_page_preview": True, }, diff --git a/src/utils/config_loader.py b/src/utils/config_loader.py index 5fbcd5c4..f151bd55 100644 --- a/src/utils/config_loader.py +++ b/src/utils/config_loader.py @@ -1,6 +1,11 @@ import os from dotenv import load_dotenv +from src.utils.telegram_chat_ids import ( + get_primary_telegram_chat_id_from_env, + get_telegram_chat_ids_from_env, +) + def load_config(): """ @@ -23,7 +28,8 @@ def load_config(): }, "telegram": { "bot_token": os.getenv("TELEGRAM_BOT_TOKEN"), - "chat_id": os.getenv("TELEGRAM_CHAT_ID"), + "chat_id": get_primary_telegram_chat_id_from_env(), + "chat_ids": get_telegram_chat_ids_from_env(), "proxy": os.getenv("HTTPS_PROXY") or os.getenv("HTTP_PROXY"), }, "config": { diff --git a/src/utils/telegram_chat_ids.py b/src/utils/telegram_chat_ids.py new file mode 100644 index 00000000..77261d54 --- /dev/null +++ b/src/utils/telegram_chat_ids.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +import os +from typing import List + + +def _split_chat_ids(raw: str | None) -> List[str]: + if not raw: + return [] + normalized = str(raw).replace("\r", ",").replace("\n", ",").replace(";", ",") + out: List[str] = [] + for token in normalized.split(","): + value = token.strip() + if value: + out.append(value) + return out + + +def parse_telegram_chat_ids(*raw_values: str | None) -> List[str]: + """Parse and de-duplicate chat ids while keeping original order.""" + out: List[str] = [] + seen: set[str] = set() + for raw in raw_values: + for chat_id in _split_chat_ids(raw): + if chat_id in seen: + continue + seen.add(chat_id) + out.append(chat_id) + return out + + +def get_telegram_chat_ids_from_env() -> List[str]: + """ + Preferred env is TELEGRAM_CHAT_IDS (comma-separated). + TELEGRAM_CHAT_ID is kept for backward compatibility. + """ + return parse_telegram_chat_ids( + os.getenv("TELEGRAM_CHAT_IDS"), + os.getenv("TELEGRAM_CHAT_ID"), + ) + + +def get_primary_telegram_chat_id_from_env() -> str: + chat_ids = get_telegram_chat_ids_from_env() + return chat_ids[0] if chat_ids else "" diff --git a/src/utils/telegram_push.py b/src/utils/telegram_push.py index f581761a..1699a309 100644 --- a/src/utils/telegram_push.py +++ b/src/utils/telegram_push.py @@ -10,6 +10,7 @@ from typing import Any, Dict, List, Optional, Tuple from loguru import logger from src.data_collection.city_registry import CITY_REGISTRY +from src.utils.telegram_chat_ids import get_telegram_chat_ids_from_env SEVERITY_RANK = { @@ -516,7 +517,7 @@ def build_trade_alert_for_city( def _maybe_send_alert( bot: Any, - chat_id: str, + chat_ids: List[str], city: str, alert_payload: Dict[str, Any], state: Dict[str, Any], @@ -552,6 +553,9 @@ def _maybe_send_alert( 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") @@ -568,7 +572,21 @@ def _maybe_send_alert( if last_sig_ts and now_ts - last_sig_ts < cooldown_sec: return False - bot.send_message(chat_id, message) + 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( + "trade alert 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, @@ -581,19 +599,19 @@ def _maybe_send_alert( logger.info( f"trade alert 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)}" + f"evidence={_evidence_brief(alert_payload)} chat_targets={sent_count}" ) return True def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[threading.Thread]: enabled = _env_bool("TELEGRAM_ALERT_PUSH_ENABLED", True) - chat_id = os.getenv("TELEGRAM_CHAT_ID") + chat_ids = get_telegram_chat_ids_from_env() if not enabled: logger.info("telegram alert push loop disabled") return None - if not chat_id: - logger.warning("telegram alert push loop skipped: TELEGRAM_CHAT_ID is not set") + if not chat_ids: + logger.warning("telegram alert push loop skipped: TELEGRAM_CHAT_IDS is not set") return None mispricing_only = _env_bool("TELEGRAM_ALERT_MISPRICING_ONLY", True) @@ -616,7 +634,7 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th logger.exception(f"failed to initialize telegram push state path={state_path}") logger.info( f"telegram alert push loop started mode={'mispricing-only' if mispricing_only else 'full'} " - f"cities={len(cities)} interval={interval_sec}s " + f"cities={len(cities)} interval={interval_sec}s chat_targets={len(chat_ids)} " f"cooldown={cooldown_sec}s min_triggers={min_trigger_count} min_severity={min_severity} " f"state_path={state_path}" ) @@ -630,7 +648,7 @@ def start_trade_alert_push_loop(bot: Any, config: Dict[str, Any]) -> Optional[th alert_payload = build_trade_alert_for_city(city, config) if _maybe_send_alert( bot=bot, - chat_id=chat_id, + chat_ids=chat_ids, city=city, alert_payload=alert_payload, state=state,