ManualTrader/utils/strategy_bot.py

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