""" Main entry point for live crypto trading bot. This module provides the main trading loop that connects to an exchange API (when implemented) and executes the Fibonacci averaging strategy. """ import time import signal import sys from datetime import datetime from typing import Optional from config import ( RISK_PERCENT, BASE_MULTIPLIER, LOG_LEVEL, LOG_FILE ) from normalization import get_spread, normalize_price, get_pip_size from thread_manager import ThreadManager, ThreadStatus from order_engine import OrderEngine from risk_manager import RiskManager from averaging_engine import AveragingEngine from exchange_api import ExchangeAPI, create_exchange from utils import ( setup_logging, get_logger, log_trade, log_balance, log_thread_event ) class TradingBot: """ Main trading bot class. Orchestrates the trading logic by connecting to an exchange, managing threads, and executing the averaging strategy. """ def __init__( self, symbol: str = "BTCUSDT", exchange_name: str = "binance", api_key: str = "", api_secret: str = "", testnet: bool = True, risk_percent: float = RISK_PERCENT, base_multiplier: float = BASE_MULTIPLIER ): """ Initialize trading bot. Args: symbol: Trading pair symbol exchange_name: Exchange to use (binance, etc.) api_key: Exchange API key api_secret: Exchange API secret testnet: Use testnet if True risk_percent: Risk percentage per trade base_multiplier: Base multiplier for TP/entry distance """ self.symbol = symbol self.risk_percent = risk_percent self.base_multiplier = base_multiplier # Setup logging self.logger = setup_logging( level=LOG_LEVEL, log_file=LOG_FILE, console=True, file=True ) # Initialize exchange connection self.exchange: ExchangeAPI = create_exchange( exchange_name, api_key=api_key, api_secret=api_secret, testnet=testnet ) # Initialize components self.thread_manager = ThreadManager() self.risk_manager = RiskManager(risk_percent=risk_percent) self.averaging_engine = AveragingEngine( thread_manager=self.thread_manager, risk_manager=self.risk_manager, symbol=symbol ) # State self.running = False self.last_price = 0.0 self.last_bid = 0.0 self.last_ask = 0.0 # Setup signal handlers signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGTERM, self._signal_handler) def _signal_handler(self, signum, frame): """Handle shutdown signals.""" self.logger.info("Shutdown signal received") self.running = False def _get_current_price(self) -> tuple: """ Get current bid/ask prices from exchange. Returns: Tuple of (bid, ask) prices """ ticker = self.exchange.get_ticker(self.symbol) if ticker: self.last_bid = ticker["bid"] self.last_ask = ticker["ask"] self.last_price = (ticker["bid"] + ticker["ask"]) / 2 return (ticker["bid"], ticker["ask"]) return (self.last_bid, self.last_ask) def _calculate_trading_parameters( self, bid: float, ask: float ) -> tuple: """ Calculate TP and entry distances based on spread. Args: bid: Current bid price ask: Current ask price Returns: Tuple of (tp_distance, entry_distance) """ spread = get_spread(bid, ask) tp_distance = spread * self.base_multiplier entry_distance = spread * self.base_multiplier return (tp_distance, entry_distance) def _get_balance(self) -> float: """Get current account balance.""" balance = self.exchange.get_balance() return balance.get("USDT", 0.0) def _should_open_new_thread(self) -> bool: """ Check if a new thread should be opened. Returns: True if new thread should be opened """ active_threads = self.thread_manager.get_active_threads() return len(active_threads) == 0 def _open_new_thread( self, bid: float, ask: float, current_time: datetime ) -> bool: """ Open a new trading thread. Args: bid: Current bid price ask: Current ask price current_time: Current timestamp Returns: True if thread was opened successfully """ balance = self._get_balance() # Calculate lot size lot_size = self.risk_manager.calculate_lot_size( balance, self.symbol, ask ) # Calculate TP and entry distances tp_distance, entry_distance = self._calculate_trading_parameters(bid, ask) tp_price = ask + tp_distance # Place buy order order = self.exchange.place_market_buy( self.symbol, lot_size ) if order and order.get("status") == "filled": fill_price = order.get("price", ask) # Create thread thread = self.thread_manager.create_thread( symbol=self.symbol, entry_price=fill_price, entry_time=current_time, lot_size=lot_size, tp_distance=tp_distance, entry_distance=entry_distance, order_id=order.get("order_id", 0) ) log_trade( "BUY", self.symbol, fill_price, lot_size, thread.magic_number, tp=tp_price ) self.logger.info(f"Opened new thread {thread.magic_number}") return True return False def _process_averaging( self, thread, current_price: float, current_time: datetime ) -> None: """ Process averaging logic for a thread. Args: thread: Thread to process current_price: Current market price current_time: Current timestamp """ # Check if averaging is needed avg_level = self.averaging_engine.check_averaging_needed( thread, current_price, is_buy=True ) if avg_level: # Place averaging order order = self.exchange.place_market_buy( self.symbol, avg_level.lot_size ) if order and order.get("status") == "filled": fill_price = order.get("price", current_price) self.thread_manager.add_averaging_order( thread.magic_number, fill_price, avg_level.lot_size, avg_level.tp_price, current_time, avg_level.level, order.get("order_id", 0) ) log_trade( "AVERAGING", self.symbol, fill_price, avg_level.lot_size, thread.magic_number, tp=avg_level.tp_price ) def _check_tp_hits( self, current_price: float, current_time: datetime ) -> None: """ Check and process TP hits for all threads. Args: current_price: Current market price current_time: Current timestamp """ for thread in self.thread_manager.get_active_threads(): for order in thread.get_open_orders(): if self.averaging_engine.check_tp_hit( order, current_price, is_buy=True ): # Close position result = self.exchange.place_market_sell( self.symbol, order.lots ) if result and result.get("status") == "filled": close_price = result.get("price", current_price) profit = self.thread_manager.close_order( thread.magic_number, order.order_id, close_price, current_time ) log_trade( "TP_HIT", self.symbol, close_price, order.lots, thread.magic_number, profit=profit ) # Handle TP cascade if order.is_averaging: self.averaging_engine.handle_tp_cascade( thread, order.averaging_level, current_price, current_time, self.exchange, is_buy=True ) # Check if thread is complete if not thread.get_open_orders(): self.thread_manager.close_thread(thread.magic_number) def _process_pending_reentries( self, current_price: float, current_time: datetime ) -> None: """ Process any pending re-entries. Args: current_price: Current market price current_time: Current timestamp """ for thread in self.thread_manager.threads.values(): if thread.pending_reentry: self.averaging_engine.process_pending_reentry( thread, current_price, current_time, self.exchange, is_buy=True ) def _log_status(self) -> None: """Log current status.""" balance = self._get_balance() stats = self.thread_manager.get_statistics() log_balance( balance=balance, equity=balance, # Simplified - would need position value calc open_positions=stats["open_positions"] ) def run(self, poll_interval: float = 1.0) -> None: """ Run the main trading loop. Args: poll_interval: Seconds between price checks """ self.logger.info(f"Starting trading bot for {self.symbol}") self.logger.info(f"Risk: {self.risk_percent}% | Multiplier: {self.base_multiplier}") self.running = True iteration = 0 while self.running: try: current_time = datetime.now() # Get current price bid, ask = self._get_current_price() if bid == 0 or ask == 0: self.logger.warning("Failed to get price, retrying...") time.sleep(poll_interval) continue # Check TP hits self._check_tp_hits(bid, current_time) # Process pending re-entries self._process_pending_reentries(bid, current_time) # Check for new thread opening if self._should_open_new_thread(): self._open_new_thread(bid, ask, current_time) # Process averaging for active threads for thread in self.thread_manager.get_active_threads(): self._process_averaging(thread, bid, current_time) # Periodic status logging iteration += 1 if iteration % 60 == 0: # Log every 60 iterations self._log_status() time.sleep(poll_interval) except Exception as e: self.logger.error(f"Error in main loop: {e}") time.sleep(poll_interval * 5) # Wait longer on error self.logger.info("Trading bot stopped") self._shutdown() def _shutdown(self) -> None: """Perform graceful shutdown.""" self.logger.info("Shutting down...") # Log final status self._log_status() # Log thread statistics stats = self.thread_manager.get_statistics() self.logger.info(f"Final Statistics: {stats}") def main(): """Main entry point.""" import argparse parser = argparse.ArgumentParser( description="Fibonacci Averaging Crypto Trading Bot" ) parser.add_argument( "--symbol", "-s", default="BTCUSDT", help="Trading pair symbol (default: BTCUSDT)" ) parser.add_argument( "--exchange", "-e", default="binance", help="Exchange to use (default: binance)" ) parser.add_argument( "--testnet", "-t", action="store_true", help="Use testnet mode" ) parser.add_argument( "--risk", "-r", type=float, default=RISK_PERCENT, help=f"Risk percentage per trade (default: {RISK_PERCENT})" ) parser.add_argument( "--multiplier", "-m", type=float, default=BASE_MULTIPLIER, help=f"Base multiplier for TP/entry (default: {BASE_MULTIPLIER})" ) parser.add_argument( "--interval", "-i", type=float, default=1.0, help="Poll interval in seconds (default: 1.0)" ) parser.add_argument( "--backtest", "-b", action="store_true", help="Run sample backtest instead of live trading" ) args = parser.parse_args() if args.backtest: from backtest import run_sample_backtest run_sample_backtest() return # For live trading, require API keys from environment or config import os api_key = os.environ.get("EXCHANGE_API_KEY", "") api_secret = os.environ.get("EXCHANGE_API_SECRET", "") if not api_key or not api_secret: print("Warning: No API keys configured. Running in simulation mode.") print("Set EXCHANGE_API_KEY and EXCHANGE_API_SECRET environment variables for live trading.") args.testnet = True bot = TradingBot( symbol=args.symbol, exchange_name=args.exchange, api_key=api_key, api_secret=api_secret, testnet=args.testnet, risk_percent=args.risk, base_multiplier=args.multiplier ) bot.run(poll_interval=args.interval) if __name__ == "__main__": main()