Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """ | |
| Telegram Notifier — sends trade and error alerts via the Telegram Bot API. | |
| Usage: | |
| notifier = TelegramNotifier(bot_token="...", chat_id="...") | |
| notifier.send_trade("LONG", "TIA", entry=5.1234, sl=4.9, tp=5.5, sz=10.0, notional=51.23) | |
| notifier.send_exit("TIA", "TP hit @ 5.5", pnl_pct=7.2) | |
| notifier.send_error("autonomous_bot", "Connection failed") | |
| Environment variables (fallback if not passed directly): | |
| TELEGRAM_BOT_TOKEN | |
| TELEGRAM_CHAT_ID | |
| """ | |
| import logging | |
| import os | |
| import threading | |
| import time | |
| from typing import Optional | |
| import requests | |
| import urllib3 | |
| try: | |
| import certifi | |
| _DEFAULT_CA = certifi.where() | |
| except ImportError: | |
| _DEFAULT_CA = True | |
| logger = logging.getLogger(__name__) | |
| class TelegramNotifier: | |
| BASE_URL = "" | |
| def __init__( | |
| self, | |
| bot_token: Optional[str] = None, | |
| chat_id: Optional[str] = None, | |
| disable_on_error: bool = False, | |
| max_consecutive_failures: int = 10, | |
| ): | |
| # Strip empty strings so env-var fallback always kicks in | |
| self.BASE_URL = os.environ.get("PROXY_URL", "").strip() | |
| self.bot_token = (bot_token or "").strip() or os.environ.get("TELEGRAM_BOT_TOKEN", "").strip() | |
| self.chat_id = (chat_id or "").strip() or os.environ.get("TELEGRAM_CHAT_ID", "").strip() | |
| self.disable_on_error = disable_on_error | |
| self.max_consecutive_failures = max_consecutive_failures | |
| self._consecutive_failures = 0 | |
| self._enabled = bool(self.bot_token and self.chat_id) | |
| self._verify: object = _DEFAULT_CA # may be flipped to False on SSL error | |
| if not self._enabled: | |
| logger.warning( | |
| "TelegramNotifier: no token/chat_id configured — notifications disabled. " | |
| "Set TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID env vars or pass them directly." | |
| ) | |
| elif not self.BASE_URL: | |
| logger.warning( | |
| "TelegramNotifier: token/chat configured but PROXY_URL is missing — messages cannot be delivered." | |
| ) | |
| else: | |
| logger.info("TelegramNotifier: configured and enabled.") | |
| # ── Public helpers ──────────────────────────────────────────────────────── | |
| def send_trade( | |
| self, | |
| direction: str, # "LONG" | "SHORT" | |
| coin: str, | |
| entry: float, | |
| sl: float, | |
| tp: float, | |
| sz: float, | |
| notional: float, | |
| strategy: str = "", | |
| timeframe: str = "", | |
| rr: Optional[float] = None, | |
| order_type: str = "", | |
| ): | |
| direction_emoji = "🟢" if direction.upper() == "LONG" else "🔴" | |
| lines = [ | |
| f"{direction_emoji} *{direction} ENTRY* — `{coin}`", | |
| f"Strategy : {strategy} {timeframe}".strip(), | |
| f"Order : {order_type}" if order_type else "", | |
| f"Entry : `{entry:.5g}`", | |
| f"SL : `{sl:.5g}`", | |
| f"TP : `{tp:.5g}`", | |
| f"Size : `{sz:.4g}` (notional ${notional:.2f})", | |
| ] | |
| if rr is not None: | |
| lines.append(f"R:R : 1:{rr:.2f}") | |
| self._send("\n".join(l for l in lines if l)) | |
| def send_exit( | |
| self, | |
| coin: str, | |
| reason: str, | |
| pnl_pct: Optional[float] = None, | |
| pnl_usdc: Optional[float] = None, | |
| strategy: str = "", | |
| timeframe: str = "", | |
| ): | |
| if pnl_usdc is not None: | |
| emoji = "✅" if pnl_usdc >= 0 else "❌" | |
| pnl_str = f"PnL: `{pnl_usdc:+.2f} USDC`" | |
| if pnl_pct is not None: | |
| pnl_str += f" (`{pnl_pct:+.2f}%`)" | |
| elif pnl_pct is not None: | |
| emoji = "✅" if pnl_pct >= 0 else "❌" | |
| pnl_str = f"PnL: `{pnl_pct:+.2f}%`" | |
| else: | |
| emoji = "🔔" | |
| pnl_str = "" | |
| lines = [ | |
| f"{emoji} *EXIT* — `{coin}`", | |
| f"Strategy : {strategy} {timeframe}".strip(), | |
| f"Reason : {reason}", | |
| ] | |
| if pnl_str: | |
| lines.append(pnl_str) | |
| self._send("\n".join(l for l in lines if l)) | |
| def send_error(self, source: str, message: str): | |
| text = f"⚠️ *ERROR* in `{source}`\n```\n{message[:800]}\n```" | |
| self._send(text) | |
| def send_info(self, message: str): | |
| self._send(f"ℹ️ {message}") | |
| # ── Internal ────────────────────────────────────────────────────────────── | |
| def _send(self, text: str, retries: int = 3): | |
| if not self._enabled: | |
| return | |
| t = threading.Thread(target=self._send_sync, args=(text, retries), daemon=True) | |
| t.start() | |
| def _send_sync(self, text: str, retries: int = 3): | |
| # Send to proxy URL (Google Apps Script) which forwards to Telegram | |
| # HuggingFace Spaces cannot access Telegram API directly, so we use the proxy | |
| url = self.BASE_URL | |
| payload = { | |
| "token": self.bot_token, | |
| "chat_id": self.chat_id, | |
| "text": text, | |
| "parse_mode": "Markdown", | |
| } | |
| headers = { | |
| "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" | |
| } | |
| if not url: | |
| logger.error("PROXY_URL not set - cannot send Telegram messages from HuggingFace Spaces") | |
| self._consecutive_failures += 1 | |
| return | |
| attempt = 0 | |
| while attempt < retries: | |
| try: | |
| resp = requests.post(url, json=payload, headers=headers, timeout=30, verify=self._verify) | |
| resp.raise_for_status() | |
| self._consecutive_failures = 0 | |
| logger.debug("Telegram message sent successfully via proxy.") | |
| return | |
| except requests.exceptions.SSLError as e: | |
| # Disable SSL verification and retry the same attempt (don't increment) | |
| if self._verify is not False: | |
| logger.warning(f"Telegram SSL error — retrying without SSL verification.") | |
| urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) | |
| self._verify = False | |
| # don't increment attempt — retry immediately with verify=False | |
| else: | |
| # Already disabled, treat as regular failure | |
| logger.warning(f"Telegram send failed (attempt {attempt+1}/{retries}): {e}") | |
| attempt += 1 | |
| if attempt < retries: | |
| time.sleep(2 ** (attempt - 1)) | |
| except Exception as e: | |
| logger.warning(f"Telegram send failed (attempt {attempt+1}/{retries}): {e}") | |
| attempt += 1 | |
| if attempt < retries: | |
| time.sleep(2 ** (attempt - 1)) | |
| # All retries exhausted | |
| self._consecutive_failures += 1 | |
| if self.disable_on_error and self._consecutive_failures >= self.max_consecutive_failures: | |
| logger.warning( | |
| f"TelegramNotifier: {self._consecutive_failures} consecutive failures — disabling." | |
| ) | |
| self._enabled = False | |
| else: | |
| logger.warning( | |
| f"TelegramNotifier: message dropped " | |
| f"({self._consecutive_failures} consecutive failure(s)). Will keep trying." | |
| ) | |