#!/usr/bin/env python3 """ Core Telegram Bot - Handles only bot setup, authentication, and basic messaging. """ import asyncio import logging from datetime import datetime from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, ContextTypes, CommandHandler, CallbackQueryHandler, MessageHandler, filters from src.config.config import Config from src.trading.trading_engine import TradingEngine from src.monitoring.market_monitor import MarketMonitor from src.notifications.notification_manager import NotificationManager from src.commands.trading_commands import TradingCommands from src.commands.info_commands import InfoCommands from src.commands.management_commands import ManagementCommands logger = logging.getLogger(__name__) class TelegramTradingBot: """Core Telegram bot handling only authentication, messaging, and command routing.""" def __init__(self): """Initialize the core bot with minimal responsibilities.""" # Core bot attributes self.application = None self.version = "Unknown" # Initialize subsystems self.trading_engine = TradingEngine() self.market_monitor = MarketMonitor(self.trading_engine) self.notification_manager = NotificationManager() # Connect notification manager to market monitor self.market_monitor.set_notification_manager(self.notification_manager) # Initialize command handlers self.trading_commands = TradingCommands(self.trading_engine, self.notification_manager) self.info_commands = InfoCommands(self.trading_engine) self.management_commands = ManagementCommands(self.trading_engine, self.market_monitor) def is_authorized(self, chat_id: str) -> bool: """Check if the chat ID is authorized to use the bot.""" return str(chat_id) == str(Config.TELEGRAM_CHAT_ID) async def send_message(self, text: str, parse_mode: str = 'HTML') -> None: """Send a message to the authorized chat.""" if self.application and Config.TELEGRAM_CHAT_ID: try: await self.application.bot.send_message( chat_id=Config.TELEGRAM_CHAT_ID, text=text, parse_mode=parse_mode ) except Exception as e: logger.error(f"Failed to send message: {e}") def setup_handlers(self): """Set up command handlers for the bot.""" if not self.application: return # Basic bot commands self.application.add_handler(CommandHandler("start", self.start_command)) self.application.add_handler(CommandHandler("help", self.help_command)) # Trading commands self.application.add_handler(CommandHandler("long", self.trading_commands.long_command)) self.application.add_handler(CommandHandler("short", self.trading_commands.short_command)) self.application.add_handler(CommandHandler("exit", self.trading_commands.exit_command)) self.application.add_handler(CommandHandler("sl", self.trading_commands.sl_command)) self.application.add_handler(CommandHandler("tp", self.trading_commands.tp_command)) self.application.add_handler(CommandHandler("coo", self.trading_commands.coo_command)) # Info commands self.application.add_handler(CommandHandler("balance", self.info_commands.balance_command)) self.application.add_handler(CommandHandler("positions", self.info_commands.positions_command)) self.application.add_handler(CommandHandler("orders", self.info_commands.orders_command)) self.application.add_handler(CommandHandler("stats", self.info_commands.stats_command)) self.application.add_handler(CommandHandler("trades", self.info_commands.trades_command)) self.application.add_handler(CommandHandler("market", self.info_commands.market_command)) self.application.add_handler(CommandHandler("price", self.info_commands.price_command)) self.application.add_handler(CommandHandler("performance", self.info_commands.performance_command)) self.application.add_handler(CommandHandler("daily", self.info_commands.daily_command)) self.application.add_handler(CommandHandler("weekly", self.info_commands.weekly_command)) self.application.add_handler(CommandHandler("monthly", self.info_commands.monthly_command)) self.application.add_handler(CommandHandler("risk", self.info_commands.risk_command)) self.application.add_handler(CommandHandler("balance_adjustments", self.info_commands.balance_adjustments_command)) self.application.add_handler(CommandHandler("commands", self.info_commands.commands_command)) self.application.add_handler(CommandHandler("c", self.info_commands.commands_command)) # Alias # Management commands self.application.add_handler(CommandHandler("monitoring", self.management_commands.monitoring_command)) self.application.add_handler(CommandHandler("alarm", self.management_commands.alarm_command)) self.application.add_handler(CommandHandler("logs", self.management_commands.logs_command)) self.application.add_handler(CommandHandler("debug", self.management_commands.debug_command)) self.application.add_handler(CommandHandler("version", self.management_commands.version_command)) self.application.add_handler(CommandHandler("keyboard", self.management_commands.keyboard_command)) # Callback and message handlers self.application.add_handler(CallbackQueryHandler(self.trading_commands.button_callback)) self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_keyboard_message)) async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /start command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("❌ Unauthorized access.") return welcome_text = f""" 🤖 Manual Trading Bot v{self.version} 🚀 Welcome to your personal trading assistant! 📈 Trading Commands: • /long [token] [amount] - Open long position • /short [token] [amount] - Open short position • /exit [token] - Close position • /sl [token] [price] - Set stop loss • /tp [token] [price] - Set take profit 📊 Info Commands: • /balance - Check account balance • /positions - View open positions • /stats - Trading statistics • /market [token] - Market data ⚙️ Management: • /monitoring - Toggle monitoring • /alarm - Set price alerts • /help - Full command list 🔗 Network: {Config.HYPERLIQUID_TESTNET and "Testnet" or "Mainnet"} 📊 Default Token: {Config.DEFAULT_TRADING_TOKEN} Ready to trade! Use the commands above or tap /help for more options. """ await update.message.reply_text(welcome_text, parse_mode='HTML') async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /help command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("❌ Unauthorized access.") return help_text = """ 📖 Complete Command Reference 🔄 Trading Commands: • /long [token] [USDC] [price] [sl:price] - Open long position • /short [token] [USDC] [price] [sl:price] - Open short position • /exit [token] - Close position (market order) • /sl [token] [price] - Set stop loss order • /tp [token] [price] - Set take profit order • /coo [token] - Cancel all open orders 📊 Account Info: • /balance - Account balance and equity • /positions - Open positions with P&L • /orders - Active orders • /trades - Recent trade history • /stats - Comprehensive trading statistics 📈 Market Data: • /market [token] - Market data and orderbook • /price [token] - Quick price check ⚙️ Management: • /monitoring - View/toggle order monitoring • /alarm [token] [price] - Set price alerts • /logs - View recent bot logs • /debug - Bot internal state (troubleshooting) For support or issues, check the logs or contact the administrator. """ await update.message.reply_text(help_text, parse_mode='HTML') async def handle_keyboard_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle messages from custom keyboard buttons.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("❌ Unauthorized access.") return # For now, just ignore keyboard messages or implement basic ones message_text = update.message.text.lower() # Basic keyboard shortcuts if message_text in ['balance', 'positions', 'orders', 'stats']: # Route to appropriate command if message_text == 'balance': await self.info_commands.balance_command(update, context) elif message_text == 'positions': await self.info_commands.positions_command(update, context) elif message_text == 'orders': await self.info_commands.orders_command(update, context) elif message_text == 'stats': await self.info_commands.stats_command(update, context) async def run(self): """Run the Telegram bot.""" if not Config.TELEGRAM_BOT_TOKEN: logger.error("❌ TELEGRAM_BOT_TOKEN not configured") return if not Config.TELEGRAM_CHAT_ID: logger.error("❌ TELEGRAM_CHAT_ID not configured") return try: # Create application self.application = Application.builder().token(Config.TELEGRAM_BOT_TOKEN).build() # Connect notification manager to the bot application self.notification_manager.set_bot_application(self.application) # Set up handlers self.setup_handlers() logger.info("🚀 Starting Telegram trading bot...") # Initialize the application await self.application.initialize() # Send startup notification await self.send_message( f"🤖 Manual Trading Bot v{self.version} Started\n\n" f"✅ Connected to Hyperliquid {'Testnet' if Config.HYPERLIQUID_TESTNET else 'Mainnet'}\n" f"📊 Default Symbol: {Config.DEFAULT_TRADING_TOKEN}\n" f"🔄 All systems ready!\n\n" "Use /start for quick actions or /help for all commands." ) # Start the application await self.application.start() # Start subsystems await self.market_monitor.start() # Start polling for updates logger.info("🔄 Starting update polling...") last_update_id = 0 while True: try: updates = await self.application.bot.get_updates( offset=last_update_id + 1, timeout=30, allowed_updates=None ) for update in updates: last_update_id = update.update_id await self.application.process_update(update) except Exception as e: logger.error(f"Error processing updates: {e}") await asyncio.sleep(5) except asyncio.CancelledError: logger.info("🛑 Bot polling cancelled") raise except Exception as e: logger.error(f"❌ Error in telegram bot: {e}") raise finally: # Clean shutdown try: await self.market_monitor.stop() if self.application: await self.application.stop() await self.application.shutdown() except Exception as e: logger.error(f"Error during shutdown: {e}")