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