#!/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." )