#!/usr/bin/env python3 """ Hyperliquid Manual Trading Bot - Main Launcher A simple launcher for systemd deployment. This is the ONLY file you need to run - it handles everything. """ import sys import os import asyncio import logging from datetime import datetime from pathlib import Path # Bot version BOT_VERSION = "2.4.238" # Add src directory to Python path sys.path.insert(0, str(Path(__file__).parent / "src")) try: from src.config.config import Config from src.bot.core import TelegramTradingBot from src.stats import TradingStats except ImportError as e: print(f"❌ Import error: {e}") print("💡 Make sure you're in the correct directory and dependencies are installed") sys.exit(1) # Global variables for graceful shutdown bot_instance = None class BotManager: """Manages the trading bot with simple startup and shutdown.""" def __init__(self): self.bot = None self.setup_logging() # Ensure logs directory exists os.makedirs("logs", exist_ok=True) def setup_logging(self): """Set up comprehensive logging.""" # Create logs directory os.makedirs("logs", exist_ok=True) # Set up file logging log_file = f"logs/trading_bot_{datetime.now().strftime('%Y%m%d')}.log" try: from src.config.logging_config import setup_logging setup_logging() self.logger = logging.getLogger(__name__) self.logger.info(f"Logging initialized - Log file: {log_file}") except Exception as e: # Fallback logging setup logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file), logging.StreamHandler() ] ) self.logger = logging.getLogger(__name__) self.logger.warning(f"Failed to setup advanced logging, using basic setup: {e}") def print_banner(self): """Print startup banner.""" # Get network and database status network_status = 'Testnet' if Config.HYPERLIQUID_TESTNET else '🚨 MAINNET 🚨' db_path = "data/trading_stats.sqlite" db_status = "✅ Found" if os.path.exists(db_path) else "❌ Not Found" banner = f""" \033[1;35m +-------------------------------------------------------------+ | | | 📱 H Y P E R L I Q U I D T R A D I N G B O T 📱 | | | | Version {BOT_VERSION} | | | +-------------------------------------------------------------+ \033[0m \033[1;34m▶ System Status\033[0m +--------------------+----------------------------------------+ | \033[1;36mStart Time\033[0m | {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | | \033[1;36mWorking Dir\033[0m | {os.getcwd()} | | \033[1;36mNetwork\033[0m | {network_status} | | \033[1;36mDatabase\033[0m | {db_status} at {db_path} | +--------------------+----------------------------------------+ \033[1;34m▶ Key Features\033[0m +-------------------------------------------------------------+ | | | 🤖 Control your trades from your phone via Telegram | | 📊 Get comprehensive, persistent trading statistics | | 🛡️ Managed as a systemd service for reliability | | 💾 Data is persistent between restarts | | 📱 Enjoy a professional, mobile-friendly interface | | | +-------------------------------------------------------------+ """ print(banner) def validate_configuration(self): """Validate bot configuration.""" self.logger.info("🔍 Validating configuration...") # Define required configuration attributes required_configs = { "HYPERLIQUID_WALLET_ADDRESS": "Your Hyperliquid wallet address", "TELEGRAM_BOT_TOKEN": "Your Telegram bot token", "TELEGRAM_CHAT_ID": "Your Telegram chat ID", "TELEGRAM_ENABLED": "Must be set to true to enable Telegram features" } missing_items = [] # Check for missing attributes in the Config class for attr, description in required_configs.items(): if not getattr(Config, attr, None): missing_items.append(f"- {attr}: {description}") # If there are missing items, log and print an error if missing_items: error_message = "❌ Missing or invalid configuration:" detailed_errors = "\n".join(missing_items) self.logger.error(f"{error_message}\n{detailed_errors}") # Print a user-friendly guide for fixing the configuration print(f"\n{error_message}") print(detailed_errors) print("\n💡 To fix this, please follow these steps:") print(" 1. Copy the example config: cp config/env.example .env") print(" 2. For Telegram setup, run: python utils/get_telegram_chat_id.py") print(" 3. Edit the .env file with your credentials.") print(" 4. For more help, see the SETUP_GUIDE.md.") return False self.logger.info("✅ Configuration validation passed") return True def check_stats_persistence(self): """Check and report on statistics persistence.""" self.logger.info("📊 Checking for persistent statistics...") db_file = "data/trading_stats.sqlite" if not os.path.exists(db_file): self.logger.warning(f"⚠️ No database found at {db_file}. New statistics will be tracked from scratch.") print(f"📊 No persistent database found. A new one will be created at {db_file}.") return try: # Attempt to connect and retrieve basic stats stats = TradingStats() basic_stats = stats.get_basic_stats() total_trades = basic_stats.get('total_trades', 0) start_date = basic_stats.get('start_date', 'N/A') # Log and print success message success_message = f"✅ Found existing database with {total_trades} trades since {start_date}." self.logger.info(success_message) print(success_message) except Exception as e: # Log and print a detailed error message if the database is corrupt or inaccessible error_message = f"❌ Error loading stats from {db_file}: {e}" self.logger.error(error_message, exc_info=True) print(error_message) print(" - The database file might be corrupted.") print(" - Consider backing up and deleting the file to start fresh.") async def run_bot(self): """Run the main bot.""" try: self.logger.info("🤖 Creating TelegramTradingBot instance...") self.bot = TelegramTradingBot() # Set version for bot to use in messages self.bot.version = BOT_VERSION self.logger.info("🚀 Starting Telegram bot...") # Run the bot await self.bot.run() except KeyboardInterrupt: self.logger.info("👋 Bot stopped by user (Ctrl+C in BotManager.run_bot)") except Exception as e: self.logger.error(f"❌ Bot error in BotManager.run_bot: {e}", exc_info=True) # Ensure cleanup is attempted if self.bot and self.bot.application and self.bot.application._running: self.logger.info("Attempting to stop application due to error in run_bot...") await self.bot.application.stop() raise # Re-raise the exception to be caught by main() async def start(self): """Main entry point for BotManager.""" try: # Check stats persistence before printing the banner self.check_stats_persistence() self.print_banner() if not self.validate_configuration(): return False self.logger.info("🎯 All systems ready - starting trading bot...") print("💡 Send /start to your Telegram bot for quick actions") print("🛑 Press Ctrl+C to stop\n") await self.run_bot() return True except Exception as e: self.logger.error(f"❌ Fatal error in BotManager.start: {e}", exc_info=True) print(f"\n💥 Fatal error during bot start: {e}") return False async def main(): bot_manager = BotManager() await bot_manager.start() if __name__ == "__main__": asyncio.run(main())