"""Telegram notification client for real-time trading alerts."""
import asyncio
import logging
import time
from dataclasses import dataclass
from enum import Enum
import aiohttp
logger = logging.getLogger(__name__)
class NotificationPriority(Enum):
"""Priority levels for notifications with emoji indicators."""
LOW = ("âšī¸", "info")
MEDIUM = ("đ", "medium")
HIGH = ("â ī¸", "warning")
CRITICAL = ("đ¨", "critical")
def __init__(self, emoji: str, label: str) -> None:
self.emoji = emoji
self.label = label
class LeakyBucket:
"""Rate limiter using leaky bucket algorithm."""
def __init__(self, rate: float, capacity: int = 1) -> None:
"""
Initialize rate limiter.
Args:
rate: Maximum requests per second
capacity: Bucket capacity (burst size)
"""
self._rate = rate
self._capacity = capacity
self._tokens = float(capacity)
self._last_update = time.monotonic()
self._lock = asyncio.Lock()
async def acquire(self) -> None:
"""Wait until a token is available, then consume it."""
async with self._lock:
now = time.monotonic()
elapsed = now - self._last_update
self._tokens = min(self._capacity, self._tokens + elapsed * self._rate)
self._last_update = now
if self._tokens < 1.0:
wait_time = (1.0 - self._tokens) / self._rate
await asyncio.sleep(wait_time)
self._tokens = 0.0
else:
self._tokens -= 1.0
@dataclass
class NotificationMessage:
"""Internal notification message structure."""
priority: NotificationPriority
message: str
class TelegramClient:
"""Telegram Bot API client for sending trading notifications."""
API_BASE = "https://api.telegram.org/bot{token}"
DEFAULT_TIMEOUT = 5.0 # seconds
DEFAULT_RATE = 1.0 # messages per second
def __init__(
self,
bot_token: str | None = None,
chat_id: str | None = None,
enabled: bool = True,
rate_limit: float = DEFAULT_RATE,
) -> None:
"""
Initialize Telegram client.
Args:
bot_token: Telegram bot token from @BotFather
chat_id: Target chat ID (user or group)
enabled: Enable/disable notifications globally
rate_limit: Maximum messages per second
"""
self._bot_token = bot_token
self._chat_id = chat_id
self._enabled = enabled
self._rate_limiter = LeakyBucket(rate=rate_limit)
self._session: aiohttp.ClientSession | None = None
if not enabled:
logger.info("Telegram notifications disabled via configuration")
elif bot_token is None or chat_id is None:
logger.warning(
"Telegram notifications disabled (missing bot_token or chat_id)"
)
self._enabled = False
else:
logger.info("Telegram notifications enabled for chat_id=%s", chat_id)
def _get_session(self) -> aiohttp.ClientSession:
"""Get or create aiohttp session."""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
timeout=aiohttp.ClientTimeout(total=self.DEFAULT_TIMEOUT)
)
return self._session
async def close(self) -> None:
"""Close HTTP session."""
if self._session is not None and not self._session.closed:
await self._session.close()
async def _send_notification(self, msg: NotificationMessage) -> None:
"""
Send notification to Telegram with graceful degradation.
Args:
msg: Notification message to send
"""
if not self._enabled:
return
try:
await self._rate_limiter.acquire()
formatted_message = f"{msg.priority.emoji} {msg.message}"
url = f"{self.API_BASE.format(token=self._bot_token)}/sendMessage"
payload = {
"chat_id": self._chat_id,
"text": formatted_message,
"parse_mode": "HTML",
}
session = self._get_session()
async with session.post(url, json=payload) as resp:
if resp.status != 200:
error_text = await resp.text()
logger.error(
"Telegram API error (status=%d): %s", resp.status, error_text
)
else:
logger.debug("Telegram notification sent: %s", msg.message[:50])
except asyncio.TimeoutError:
logger.error("Telegram notification timeout")
except aiohttp.ClientError as exc:
logger.error("Telegram notification failed: %s", exc)
except Exception as exc:
logger.error("Unexpected error sending notification: %s", exc)
async def notify_trade_execution(
self,
stock_code: str,
market: str,
action: str,
quantity: int,
price: float,
confidence: float,
) -> None:
"""
Notify trade execution.
Args:
stock_code: Stock ticker symbol
market: Market name (e.g., "Korea", "United States")
action: "BUY" or "SELL"
quantity: Number of shares
price: Execution price
confidence: AI confidence level (0-100)
"""
emoji = "đĸ" if action == "BUY" else "đ´"
message = (
f"{emoji} {action}\n"
f"Symbol: {stock_code} ({market})\n"
f"Quantity: {quantity:,} shares\n"
f"Price: {price:,.2f}\n"
f"Confidence: {confidence:.0f}%"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.MEDIUM, message=message)
)
async def notify_market_open(self, market_name: str) -> None:
"""
Notify market opening.
Args:
market_name: Name of the market (e.g., "Korea", "United States")
"""
message = f"Market Open\n{market_name} trading session started"
await self._send_notification(
NotificationMessage(priority=NotificationPriority.LOW, message=message)
)
async def notify_market_close(self, market_name: str, pnl_pct: float) -> None:
"""
Notify market closing.
Args:
market_name: Name of the market
pnl_pct: Final P&L percentage for the session
"""
pnl_sign = "+" if pnl_pct >= 0 else ""
pnl_emoji = "đ" if pnl_pct >= 0 else "đ"
message = (
f"Market Close\n"
f"{market_name} trading session ended\n"
f"{pnl_emoji} P&L: {pnl_sign}{pnl_pct:.2f}%"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.LOW, message=message)
)
async def notify_circuit_breaker(
self, pnl_pct: float, threshold: float
) -> None:
"""
Notify circuit breaker activation.
Args:
pnl_pct: Current P&L percentage
threshold: Circuit breaker threshold
"""
message = (
f"CIRCUIT BREAKER TRIPPED\n"
f"P&L: {pnl_pct:.2f}% (threshold: {threshold:.1f}%)\n"
f"Trading halted for safety"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.CRITICAL, message=message)
)
async def notify_fat_finger(
self,
stock_code: str,
order_amount: float,
total_cash: float,
max_pct: float,
) -> None:
"""
Notify fat-finger protection rejection.
Args:
stock_code: Stock ticker symbol
order_amount: Attempted order amount
total_cash: Total available cash
max_pct: Maximum allowed percentage
"""
attempted_pct = (order_amount / total_cash) * 100 if total_cash > 0 else 0
message = (
f"Fat-Finger Protection\n"
f"Order rejected: {stock_code}\n"
f"Attempted: {attempted_pct:.1f}% of cash\n"
f"Max allowed: {max_pct:.0f}%\n"
f"Amount: {order_amount:,.0f} / {total_cash:,.0f}"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.HIGH, message=message)
)
async def notify_system_start(
self, mode: str, enabled_markets: list[str]
) -> None:
"""
Notify system startup.
Args:
mode: Trading mode ("paper" or "live")
enabled_markets: List of enabled market codes
"""
mode_emoji = "đ" if mode == "paper" else "đ°"
markets_str = ", ".join(enabled_markets)
message = (
f"{mode_emoji} System Started\n"
f"Mode: {mode.upper()}\n"
f"Markets: {markets_str}"
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.MEDIUM, message=message)
)
async def notify_system_shutdown(self, reason: str) -> None:
"""
Notify system shutdown.
Args:
reason: Reason for shutdown (e.g., "Normal shutdown", "Circuit breaker")
"""
message = f"System Shutdown\n{reason}"
priority = (
NotificationPriority.CRITICAL
if "circuit breaker" in reason.lower()
else NotificationPriority.MEDIUM
)
await self._send_notification(
NotificationMessage(priority=priority, message=message)
)
async def notify_error(
self, error_type: str, error_msg: str, context: str
) -> None:
"""
Notify system error.
Args:
error_type: Type of error (e.g., "Connection Error")
error_msg: Error message
context: Error context (e.g., stock code, market)
"""
message = (
f"Error: {error_type}\n"
f"Context: {context}\n"
f"Message: {error_msg[:200]}" # Truncate long errors
)
await self._send_notification(
NotificationMessage(priority=NotificationPriority.HIGH, message=message)
)