#!/usr/bin/env python3 """ Telegram Bot for Hyperliquid Trading This module provides a Telegram interface for manual Hyperliquid trading with comprehensive statistics tracking and phone-friendly controls. """ import logging import asyncio import re from datetime import datetime from typing import Optional, Dict, Any from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes, MessageHandler, filters from hyperliquid_client import HyperliquidClient from trading_stats import TradingStats from config import Config # Set up logging logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=getattr(logging, Config.LOG_LEVEL) ) logger = logging.getLogger(__name__) class TelegramTradingBot: """Telegram bot for manual trading with comprehensive statistics.""" def __init__(self): """Initialize the Telegram trading bot.""" self.client = HyperliquidClient(use_testnet=Config.HYPERLIQUID_TESTNET) self.stats = TradingStats() self.authorized_chat_id = Config.TELEGRAM_CHAT_ID self.application = None # Initialize stats with current balance self._initialize_stats() def _initialize_stats(self): """Initialize stats with current balance.""" try: balance = self.client.get_balance() if balance and balance.get('total'): # Get USDC balance as the main balance usdc_balance = float(balance['total'].get('USDC', 0)) self.stats.set_initial_balance(usdc_balance) except Exception as e: logger.error(f"Could not initialize stats: {e}") def is_authorized(self, chat_id: str) -> bool: """Check if the chat ID is authorized to use the bot.""" return str(chat_id) == str(self.authorized_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 self.authorized_chat_id: try: await self.application.bot.send_message( chat_id=self.authorized_chat_id, text=text, parse_mode=parse_mode ) except Exception as e: logger.error(f"Failed to send message: {e}") 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 = """ šŸ¤– Hyperliquid Manual Trading Bot Welcome to your personal trading assistant! Control your Hyperliquid account directly from your phone. šŸ“± Quick Actions: Tap the buttons below for instant access to key functions. šŸ’¼ Account Commands: /balance - Account balance /positions - Open positions /orders - Open orders /stats - Trading statistics šŸ“Š Market Commands: /market - Market data /price - Current price šŸ”„ Trading Commands: /buy [amount] [price] - Buy order /sell [amount] [price] - Sell order /trades - Recent trades /cancel [order_id] - Cancel order šŸ“ˆ Statistics: /stats - Full trading statistics /performance - Performance metrics /risk - Risk analysis Type /help for detailed command information. """ keyboard = [ [ InlineKeyboardButton("šŸ’° Balance", callback_data="balance"), InlineKeyboardButton("šŸ“Š Stats", callback_data="stats") ], [ InlineKeyboardButton("šŸ“ˆ Positions", callback_data="positions"), InlineKeyboardButton("šŸ“‹ Orders", callback_data="orders") ], [ InlineKeyboardButton("šŸ’µ Price", callback_data="price"), InlineKeyboardButton("šŸ“Š Market", callback_data="market") ], [ InlineKeyboardButton("šŸ”„ Recent Trades", callback_data="trades"), InlineKeyboardButton("āš™ļø Help", callback_data="help") ] ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text(welcome_text, parse_mode='HTML', reply_markup=reply_markup) 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 = """ šŸ”§ Hyperliquid Trading Bot - Complete Guide šŸ’¼ Account Management: ā€¢ /balance - Show account balance ā€¢ /positions - Show open positions ā€¢ /orders - Show open orders šŸ“Š Market Data: ā€¢ /market - Detailed market data ā€¢ /price - Quick price check šŸ”„ Manual Trading: ā€¢ /buy 0.001 50000 - Buy 0.001 BTC at $50,000 ā€¢ /sell 0.001 55000 - Sell 0.001 BTC at $55,000 ā€¢ /cancel ABC123 - Cancel order with ID ABC123 šŸ“ˆ Statistics & Analytics: ā€¢ /stats - Complete trading statistics ā€¢ /performance - Win rate, profit factor, etc. ā€¢ /risk - Sharpe ratio, drawdown, VaR ā€¢ /trades - Recent trade history āš™ļø Configuration: ā€¢ Symbol: {symbol} ā€¢ Default Amount: {amount} ā€¢ Network: {network} šŸ›”ļø Safety Features: ā€¢ All trades logged automatically ā€¢ Comprehensive performance tracking ā€¢ Real-time balance monitoring ā€¢ Risk metrics calculation šŸ“± Mobile Optimized: ā€¢ Quick action buttons ā€¢ Instant notifications ā€¢ Clean, readable layout ā€¢ One-tap commands For support, contact your bot administrator. """.format( symbol=Config.DEFAULT_TRADING_SYMBOL, amount=Config.DEFAULT_TRADE_AMOUNT, network="Testnet" if Config.HYPERLIQUID_TESTNET else "Mainnet" ) await update.message.reply_text(help_text, parse_mode='HTML') async def stats_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /stats command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return # Get current balance for stats balance = self.client.get_balance() current_balance = 0 if balance and balance.get('total'): current_balance = float(balance['total'].get('USDC', 0)) stats_message = self.stats.format_stats_message(current_balance) await update.message.reply_text(stats_message, parse_mode='HTML') async def buy_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /buy command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return try: if len(context.args) < 2: await update.message.reply_text( "āŒ Usage: /buy [amount] [price]\n" f"Example: /buy {Config.DEFAULT_TRADE_AMOUNT} 50000" ) return amount = float(context.args[0]) price = float(context.args[1]) symbol = Config.DEFAULT_TRADING_SYMBOL # Confirmation message confirmation_text = f""" šŸŸ¢ Buy Order Confirmation šŸ“Š Order Details: ā€¢ Symbol: {symbol} ā€¢ Side: BUY ā€¢ Amount: {amount} ā€¢ Price: ${price:,.2f} ā€¢ Total Value: ${amount * price:,.2f} āš ļø Are you sure you want to place this order? This will attempt to buy {amount} {symbol} at ${price:,.2f} per unit. """ keyboard = [ [ InlineKeyboardButton("āœ… Confirm Buy", callback_data=f"confirm_buy_{amount}_{price}"), InlineKeyboardButton("āŒ Cancel", callback_data="cancel_order") ] ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text(confirmation_text, parse_mode='HTML', reply_markup=reply_markup) except ValueError: await update.message.reply_text("āŒ Invalid amount or price. Please use numbers only.") except Exception as e: await update.message.reply_text(f"āŒ Error processing buy command: {e}") async def sell_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /sell command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return try: if len(context.args) < 2: await update.message.reply_text( "āŒ Usage: /sell [amount] [price]\n" f"Example: /sell {Config.DEFAULT_TRADE_AMOUNT} 55000" ) return amount = float(context.args[0]) price = float(context.args[1]) symbol = Config.DEFAULT_TRADING_SYMBOL # Confirmation message confirmation_text = f""" šŸ”“ Sell Order Confirmation šŸ“Š Order Details: ā€¢ Symbol: {symbol} ā€¢ Side: SELL ā€¢ Amount: {amount} ā€¢ Price: ${price:,.2f} ā€¢ Total Value: ${amount * price:,.2f} āš ļø Are you sure you want to place this order? This will attempt to sell {amount} {symbol} at ${price:,.2f} per unit. """ keyboard = [ [ InlineKeyboardButton("āœ… Confirm Sell", callback_data=f"confirm_sell_{amount}_{price}"), InlineKeyboardButton("āŒ Cancel", callback_data="cancel_order") ] ] reply_markup = InlineKeyboardMarkup(keyboard) await update.message.reply_text(confirmation_text, parse_mode='HTML', reply_markup=reply_markup) except ValueError: await update.message.reply_text("āŒ Invalid amount or price. Please use numbers only.") except Exception as e: await update.message.reply_text(f"āŒ Error processing sell command: {e}") async def trades_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /trades command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return recent_trades = self.stats.get_recent_trades(10) if not recent_trades: await update.message.reply_text("šŸ“ No trades recorded yet.") return trades_text = "šŸ”„ Recent Trades\n\n" for trade in reversed(recent_trades[-5:]): # Show last 5 trades timestamp = datetime.fromisoformat(trade['timestamp']).strftime('%m/%d %H:%M') side_emoji = "šŸŸ¢" if trade['side'] == 'buy' else "šŸ”“" trades_text += f"{side_emoji} {trade['side'].upper()} {trade['amount']} {trade['symbol']}\n" trades_text += f" šŸ’° ${trade['price']:,.2f} | šŸ’µ ${trade['value']:,.2f}\n" trades_text += f" šŸ“… {timestamp}\n\n" await update.message.reply_text(trades_text, parse_mode='HTML') async def balance_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /balance command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return balance = self.client.get_balance() if balance: balance_text = "šŸ’° Account Balance\n\n" total_balance = balance.get('total', {}) if total_balance: total_value = 0 for asset, amount in total_balance.items(): if float(amount) > 0: balance_text += f"šŸ’µ {asset}: {amount}\n" if asset == 'USDC': total_value += float(amount) balance_text += f"\nšŸ’¼ Total Value: ${total_value:,.2f}" # Add stats summary basic_stats = self.stats.get_basic_stats() if basic_stats['initial_balance'] > 0: pnl = total_value - basic_stats['initial_balance'] pnl_percent = (pnl / basic_stats['initial_balance']) * 100 balance_text += f"\nšŸ“Š P&L: ${pnl:,.2f} ({pnl_percent:+.2f}%)" else: balance_text += "No balance data available" else: balance_text = "āŒ Could not fetch balance data" await update.message.reply_text(balance_text, parse_mode='HTML') async def positions_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /positions command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return positions = self.client.get_positions() if positions: positions_text = "šŸ“ˆ Open Positions\n\n" open_positions = [p for p in positions if float(p.get('contracts', 0)) != 0] if open_positions: total_unrealized = 0 for position in open_positions: symbol = position.get('symbol', 'Unknown') contracts = float(position.get('contracts', 0)) unrealized_pnl = float(position.get('unrealizedPnl', 0)) entry_price = float(position.get('entryPx', 0)) pnl_emoji = "šŸŸ¢" if unrealized_pnl >= 0 else "šŸ”“" positions_text += f"šŸ“Š {symbol}\n" positions_text += f" šŸ“ Size: {contracts} contracts\n" positions_text += f" šŸ’° Entry: ${entry_price:,.2f}\n" positions_text += f" {pnl_emoji} PnL: ${unrealized_pnl:,.2f}\n\n" total_unrealized += unrealized_pnl positions_text += f"šŸ’¼ Total Unrealized P&L: ${total_unrealized:,.2f}" else: positions_text += "No open positions" else: positions_text = "āŒ Could not fetch positions data" await update.message.reply_text(positions_text, parse_mode='HTML') async def orders_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /orders command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return orders = self.client.get_open_orders() if orders: orders_text = "šŸ“‹ Open Orders\n\n" if orders and len(orders) > 0: for order in orders: symbol = order.get('symbol', 'Unknown') side = order.get('side', 'Unknown') amount = order.get('amount', 0) price = order.get('price', 0) order_id = order.get('id', 'Unknown') side_emoji = "šŸŸ¢" if side.lower() == 'buy' else "šŸ”“" orders_text += f"{side_emoji} {symbol}\n" orders_text += f" šŸ“Š {side.upper()} {amount} @ ${price:,.2f}\n" orders_text += f" šŸ’µ Value: ${float(amount) * float(price):,.2f}\n" orders_text += f" šŸ”‘ ID: {order_id}\n\n" else: orders_text += "No open orders" else: orders_text = "āŒ Could not fetch orders data" await update.message.reply_text(orders_text, parse_mode='HTML') async def market_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /market command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return symbol = Config.DEFAULT_TRADING_SYMBOL market_data = self.client.get_market_data(symbol) if market_data: ticker = market_data['ticker'] orderbook = market_data['orderbook'] # Calculate 24h change current_price = float(ticker.get('last', 0)) high_24h = float(ticker.get('high', 0)) low_24h = float(ticker.get('low', 0)) market_text = f"šŸ“Š Market Data - {symbol}\n\n" market_text += f"šŸ’µ Current Price: ${current_price:,.2f}\n" market_text += f"šŸ“ˆ 24h High: ${high_24h:,.2f}\n" market_text += f"šŸ“‰ 24h Low: ${low_24h:,.2f}\n" market_text += f"šŸ“Š 24h Volume: {ticker.get('baseVolume', 'N/A')}\n\n" if orderbook.get('bids') and orderbook.get('asks'): best_bid = float(orderbook['bids'][0][0]) if orderbook['bids'] else 0 best_ask = float(orderbook['asks'][0][0]) if orderbook['asks'] else 0 spread = best_ask - best_bid spread_percent = (spread / best_ask * 100) if best_ask > 0 else 0 market_text += f"šŸŸ¢ Best Bid: ${best_bid:,.2f}\n" market_text += f"šŸ”“ Best Ask: ${best_ask:,.2f}\n" market_text += f"šŸ“ Spread: ${spread:.2f} ({spread_percent:.3f}%)\n" else: market_text = "āŒ Could not fetch market data" await update.message.reply_text(market_text, parse_mode='HTML') async def price_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle the /price command.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return symbol = Config.DEFAULT_TRADING_SYMBOL market_data = self.client.get_market_data(symbol) if market_data: price = float(market_data['ticker'].get('last', 0)) price_text = f"šŸ’µ {symbol}: ${price:,.2f}" # Add timestamp timestamp = datetime.now().strftime('%H:%M:%S') price_text += f"\nā° Updated: {timestamp}" else: price_text = f"āŒ Could not fetch price for {symbol}" await update.message.reply_text(price_text, parse_mode='HTML') async def button_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle inline keyboard button presses.""" query = update.callback_query await query.answer() if not self.is_authorized(query.message.chat_id): await query.edit_message_text("āŒ Unauthorized access.") return callback_data = query.data # Handle trading confirmations if callback_data.startswith('confirm_buy_'): parts = callback_data.split('_') amount = float(parts[2]) price = float(parts[3]) await self._execute_buy_order(query, amount, price) return elif callback_data.startswith('confirm_sell_'): parts = callback_data.split('_') amount = float(parts[2]) price = float(parts[3]) await self._execute_sell_order(query, amount, price) return elif callback_data == 'cancel_order': await query.edit_message_text("āŒ Order cancelled.") return # Create a fake update object for reusing command handlers fake_update = Update( update_id=update.update_id, message=query.message, callback_query=query ) # Handle regular button callbacks if callback_data == "balance": await self.balance_command(fake_update, context) elif callback_data == "stats": await self.stats_command(fake_update, context) elif callback_data == "positions": await self.positions_command(fake_update, context) elif callback_data == "orders": await self.orders_command(fake_update, context) elif callback_data == "market": await self.market_command(fake_update, context) elif callback_data == "price": await self.price_command(fake_update, context) elif callback_data == "trades": await self.trades_command(fake_update, context) elif callback_data == "help": await self.help_command(fake_update, context) async def _execute_buy_order(self, query, amount: float, price: float): """Execute a buy order.""" symbol = Config.DEFAULT_TRADING_SYMBOL try: await query.edit_message_text("ā³ Placing buy order...") # Place the order order = self.client.place_limit_order(symbol, 'buy', amount, price) if order: # Record the trade in stats order_id = order.get('id', 'N/A') self.stats.record_trade(symbol, 'buy', amount, price, order_id) success_message = f""" āœ… Buy Order Placed Successfully! šŸ“Š Order Details: ā€¢ Symbol: {symbol} ā€¢ Side: BUY ā€¢ Amount: {amount} ā€¢ Price: ${price:,.2f} ā€¢ Order ID: {order_id} ā€¢ Total Value: ${amount * price:,.2f} The order has been submitted to Hyperliquid. """ await query.edit_message_text(success_message, parse_mode='HTML') logger.info(f"Buy order placed: {amount} {symbol} @ ${price}") else: await query.edit_message_text("āŒ Failed to place buy order. Please try again.") except Exception as e: error_message = f"āŒ Error placing buy order: {str(e)}" await query.edit_message_text(error_message) logger.error(f"Error placing buy order: {e}") async def _execute_sell_order(self, query, amount: float, price: float): """Execute a sell order.""" symbol = Config.DEFAULT_TRADING_SYMBOL try: await query.edit_message_text("ā³ Placing sell order...") # Place the order order = self.client.place_limit_order(symbol, 'sell', amount, price) if order: # Record the trade in stats order_id = order.get('id', 'N/A') self.stats.record_trade(symbol, 'sell', amount, price, order_id) success_message = f""" āœ… Sell Order Placed Successfully! šŸ“Š Order Details: ā€¢ Symbol: {symbol} ā€¢ Side: SELL ā€¢ Amount: {amount} ā€¢ Price: ${price:,.2f} ā€¢ Order ID: {order_id} ā€¢ Total Value: ${amount * price:,.2f} The order has been submitted to Hyperliquid. """ await query.edit_message_text(success_message, parse_mode='HTML') logger.info(f"Sell order placed: {amount} {symbol} @ ${price}") else: await query.edit_message_text("āŒ Failed to place sell order. Please try again.") except Exception as e: error_message = f"āŒ Error placing sell order: {str(e)}" await query.edit_message_text(error_message) logger.error(f"Error placing sell order: {e}") async def unknown_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle unknown commands.""" if not self.is_authorized(update.effective_chat.id): await update.message.reply_text("āŒ Unauthorized access.") return await update.message.reply_text( "ā“ Unknown command. Use /help to see available commands or tap the buttons in /start." ) def setup_handlers(self): """Set up command handlers for the bot.""" if not self.application: return # Command handlers self.application.add_handler(CommandHandler("start", self.start_command)) self.application.add_handler(CommandHandler("help", self.help_command)) self.application.add_handler(CommandHandler("balance", self.balance_command)) self.application.add_handler(CommandHandler("positions", self.positions_command)) self.application.add_handler(CommandHandler("orders", self.orders_command)) self.application.add_handler(CommandHandler("market", self.market_command)) self.application.add_handler(CommandHandler("price", self.price_command)) self.application.add_handler(CommandHandler("stats", self.stats_command)) self.application.add_handler(CommandHandler("trades", self.trades_command)) self.application.add_handler(CommandHandler("buy", self.buy_command)) self.application.add_handler(CommandHandler("sell", self.sell_command)) # Callback query handler for inline keyboards self.application.add_handler(CallbackQueryHandler(self.button_callback)) # Handle unknown commands self.application.add_handler(MessageHandler(filters.COMMAND, self.unknown_command)) 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 # Create application self.application = Application.builder().token(Config.TELEGRAM_BOT_TOKEN).build() # Set up handlers self.setup_handlers() logger.info("šŸš€ Starting Telegram trading bot...") # Send startup notification await self.send_message( "šŸ¤– Manual Trading Bot Started\n\n" f"āœ… Connected to Hyperliquid {'Testnet' if Config.HYPERLIQUID_TESTNET else 'Mainnet'}\n" f"šŸ“Š Default Symbol: {Config.DEFAULT_TRADING_SYMBOL}\n" f"šŸ“± Manual trading ready!\n" f"ā° Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" "Use /start for quick actions or /help for all commands." ) # Start the bot await self.application.run_polling() def main(): """Main entry point for the Telegram bot.""" try: # Validate configuration if not Config.validate(): logger.error("āŒ Configuration validation failed!") return if not Config.TELEGRAM_ENABLED: logger.error("āŒ Telegram is not enabled in configuration") return # Create and run the bot bot = TelegramTradingBot() asyncio.run(bot.run()) except KeyboardInterrupt: logger.info("šŸ‘‹ Bot stopped by user") except Exception as e: logger.error(f"āŒ Unexpected error: {e}") if __name__ == "__main__": main()