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