#!/usr/bin/env python3 """ Automated Trading Strategy Bot This bot runs continuously and implements automated trading strategies. It can run alongside the Telegram bot for monitoring and control. """ import asyncio import time import logging from datetime import datetime, timedelta from typing import Optional, Dict, Any from hyperliquid_client import HyperliquidClient 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 TradingStrategy: """Base class for trading strategies.""" def __init__(self, client: HyperliquidClient): self.client = client self.symbol = Config.DEFAULT_TRADING_SYMBOL self.trade_amount = Config.DEFAULT_TRADE_AMOUNT async def analyze_market(self) -> Dict[str, Any]: """Analyze market conditions. Override in subclasses.""" market_data = self.client.get_market_data(self.symbol) return { 'signal': 'HOLD', # BUY, SELL, HOLD 'confidence': 0.0, # 0.0 to 1.0 'market_data': market_data } async def execute_strategy(self) -> bool: """Execute the trading strategy. Override in subclasses.""" analysis = await self.analyze_market() logger.info(f"Strategy analysis: {analysis['signal']} (confidence: {analysis['confidence']:.2f})") return True class SimpleMovingAverageStrategy(TradingStrategy): """Example: Simple Moving Average Crossover Strategy.""" def __init__(self, client: HyperliquidClient): super().__init__(client) self.price_history = [] self.short_period = 5 # 5 minute MA self.long_period = 20 # 20 minute MA async def analyze_market(self) -> Dict[str, Any]: """Analyze using moving average crossover.""" market_data = self.client.get_market_data(self.symbol) if not market_data: return {'signal': 'HOLD', 'confidence': 0.0, 'market_data': None} current_price = float(market_data['ticker'].get('last', 0)) if current_price == 0: return {'signal': 'HOLD', 'confidence': 0.0, 'market_data': market_data} # Add current price to history self.price_history.append(current_price) # Keep only what we need max_history = max(self.short_period, self.long_period) + 5 if len(self.price_history) > max_history: self.price_history = self.price_history[-max_history:] # Need enough data for analysis if len(self.price_history) < self.long_period: return {'signal': 'HOLD', 'confidence': 0.0, 'market_data': market_data} # Calculate moving averages short_ma = sum(self.price_history[-self.short_period:]) / self.short_period long_ma = sum(self.price_history[-self.long_period:]) / self.long_period # Previous MAs for crossover detection if len(self.price_history) >= self.long_period + 1: prev_short_ma = sum(self.price_history[-self.short_period-1:-1]) / self.short_period prev_long_ma = sum(self.price_history[-self.long_period-1:-1]) / self.long_period # Bullish crossover: short MA crosses above long MA if prev_short_ma <= prev_long_ma and short_ma > long_ma: return {'signal': 'BUY', 'confidence': 0.7, 'market_data': market_data} # Bearish crossover: short MA crosses below long MA if prev_short_ma >= prev_long_ma and short_ma < long_ma: return {'signal': 'SELL', 'confidence': 0.7, 'market_data': market_data} return {'signal': 'HOLD', 'confidence': 0.5, 'market_data': market_data} async def execute_strategy(self) -> bool: """Execute the moving average strategy.""" analysis = await self.analyze_market() signal = analysis['signal'] confidence = analysis['confidence'] logger.info(f"MA Strategy: {signal} (confidence: {confidence:.2f})") # Only trade with high confidence if confidence < 0.6: return True # Get current positions to avoid overtrading positions = self.client.get_positions(self.symbol) current_position = 0 if positions: for pos in positions: if pos.get('symbol') == self.symbol: current_position = float(pos.get('contracts', 0)) break market_data = analysis['market_data'] current_price = float(market_data['ticker'].get('last', 0)) if signal == 'BUY' and current_position <= 0: # Place buy order slightly below market price buy_price = current_price * 0.999 # 0.1% below market logger.info(f"🟢 Placing BUY order: {self.trade_amount} {self.symbol} @ ${buy_price:.2f}") # Uncomment to enable real trading: # order = self.client.place_limit_order(self.symbol, 'buy', self.trade_amount, buy_price) # if order: # logger.info(f"✅ Buy order placed: {order}") # else: # logger.error("❌ Failed to place buy order") elif signal == 'SELL' and current_position >= 0: # Place sell order slightly above market price sell_price = current_price * 1.001 # 0.1% above market logger.info(f"🔴 Placing SELL order: {self.trade_amount} {self.symbol} @ ${sell_price:.2f}") # Uncomment to enable real trading: # order = self.client.place_limit_order(self.symbol, 'sell', self.trade_amount, sell_price) # if order: # logger.info(f"✅ Sell order placed: {order}") # else: # logger.error("❌ Failed to place sell order") return True class StrategyBot: """Main automated trading bot that runs strategies continuously.""" def __init__(self): """Initialize the strategy bot.""" self.client = HyperliquidClient(use_testnet=Config.HYPERLIQUID_TESTNET) self.strategies = [] self.is_running = False self.loop_interval = 60 # Run every 60 seconds # Add strategies self.add_strategy(SimpleMovingAverageStrategy(self.client)) def add_strategy(self, strategy: TradingStrategy): """Add a trading strategy to the bot.""" self.strategies.append(strategy) logger.info(f"Added strategy: {strategy.__class__.__name__}") async def run_strategies(self): """Run all strategies once.""" for strategy in self.strategies: try: await strategy.execute_strategy() except Exception as e: logger.error(f"Error in strategy {strategy.__class__.__name__}: {e}") async def health_check(self): """Check bot health and connection.""" # Test connection market_data = self.client.get_market_data(Config.DEFAULT_TRADING_SYMBOL) if not market_data: logger.error("❌ Health check failed - cannot connect to Hyperliquid") return False # Check balance balance = self.client.get_balance() if not balance: logger.warning("⚠️ Could not fetch balance") logger.info("✅ Health check passed") return True async def run(self): """Main bot loop.""" logger.info("🚀 Starting Strategy Bot...") logger.info(f"📊 Default Symbol: {Config.DEFAULT_TRADING_SYMBOL}") logger.info(f"💰 Trade Amount: {Config.DEFAULT_TRADE_AMOUNT}") logger.info(f"⏰ Loop Interval: {self.loop_interval} seconds") logger.info(f"🌐 Network: {'Testnet' if Config.HYPERLIQUID_TESTNET else 'Mainnet'}") # Initial health check if not await self.health_check(): logger.error("❌ Initial health check failed. Exiting.") return self.is_running = True iteration = 0 try: while self.is_running: iteration += 1 start_time = time.time() logger.info(f"🔄 Strategy iteration #{iteration} started") # Run strategies await self.run_strategies() # Periodic health check (every 10 iterations) if iteration % 10 == 0: await self.health_check() # Calculate sleep time execution_time = time.time() - start_time sleep_time = max(0, self.loop_interval - execution_time) logger.info(f"✅ Iteration #{iteration} completed in {execution_time:.2f}s. Sleeping {sleep_time:.2f}s") if sleep_time > 0: await asyncio.sleep(sleep_time) except KeyboardInterrupt: logger.info("👋 Strategy bot stopped by user") except Exception as e: logger.error(f"❌ Unexpected error in strategy bot: {e}") finally: self.is_running = False logger.info("🛑 Strategy bot stopped") def main(): """Main entry point.""" try: # Validate configuration if not Config.validate(): logger.error("❌ Configuration validation failed!") return # Create and run the bot bot = StrategyBot() 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()