realtime_arbitrage.py

"""Real-time arbitrage detection system with WebSocket feeds."""

import time
import signal
import sys
import os
from typing import Dict, List
from logger import logger, init_logger
from ccxt_interface import CCXTInterface
from cross_rate_matrix import CrossRateMatrix
import config
from arbitrage_trade import ArbitrageTrade


class RealTimeArbitrageDetector:
    """Main class for real-time arbitrage detection using WebSocket feeds."""
    
    def __init__(self, exchange_id: str = 'binance', limit_markets: int = 80):
        self.exchange_id = exchange_id
        self.limit_markets = limit_markets
        self.running = False
        
        # Initialize components
        logger.info("Initializing Real-Time Arbitrage Detector...")
        
        # Setup CCXT interface with polling support
        self.ccxt_interface = CCXTInterface(
            exchange_id=exchange_id,
            limit_markets=limit_markets,
            enable_live_updates=True
        )

        self.trade_executor = ArbitrageTrade(
            exchange=exchange_id,
            api_key=os.getenv('EXCHANGE_API_KEY', ''),  # Read API key from environment
            api_secret=os.getenv('EXCHANGE_API_SECRET', ''),  # Read API secret from environment
        )

        # Setup cross-rate matrix
        self.cross_rate_matrix = CrossRateMatrix()
        
        # Register for price updates
        self.ccxt_interface.add_price_callback(self._on_price_update)
        
        # Statistics
        self.update_count = 0
        self.arbitrage_opportunities = []
        self.last_arbitrage_check = 0
        self.arbitrage_check_interval = 5.0  # Check every 5 seconds
        
        logger.info("Real-Time Arbitrage Detector initialized")
    
    def _on_price_update(self, symbol: str, price_data: Dict):
        """Callback for price updates from WebSocket."""
        self.update_count += 1
        
        if self.update_count % 100 == 0:
            logger.info(f"Processed {self.update_count} price updates")
        
        # Update cross-rate matrix periodically
        current_time = time.time()
        if current_time - self.last_arbitrage_check >= self.arbitrage_check_interval:
            self._check_arbitrage_opportunities()
            self.last_arbitrage_check = current_time
    
    def _check_arbitrage_opportunities(self):
        """Check for arbitrage opportunities using current price data."""
        try:
            # Get all current prices
            all_prices = self.ccxt_interface.get_all_cached_prices()
            
            if len(all_prices) < 3:  # Need at least 3 pairs for triangular arbitrage
                logger.debug(f"Not enough price data: {len(all_prices)} symbols")
                return
            
            logger.debug(f"Checking arbitrage with {len(all_prices)} symbols")
            
            # Update cross-rate matrix
            self.cross_rate_matrix.update_from_symbol_prices(all_prices, self.ccxt_interface.markets)
            
            # Find arbitrage cycles
            cycles = self.cross_rate_matrix.find_arbitrage_cycles(
                base_currency="USDT", 
                min_profit_threshold=0.0001  # 0.01% minimum profit
            )
            
            if cycles:
                logger.info(f"Found {len(cycles)} arbitrage opportunities!")
                
                # Store and display best opportunities
                for i, cycle in enumerate(cycles[:3]):  # Show top 3
                    self._log_arbitrage_opportunity(cycle, i + 1)
                    self.trade_executor.execute_cycle(cycle, initial_amount=1, dry_run=True)
                
                self.arbitrage_opportunities = cycles
            else:
                if hasattr(self, '_no_arbitrage_logged') and not self._no_arbitrage_logged:
                    logger.info("No profitable arbitrage opportunities found")
                    self._no_arbitrage_logged = True
                    
        except Exception as e:
            import traceback
            logger.error(f"Error checking arbitrage opportunities: {e}")
            logger.debug(f"Traceback: {traceback.format_exc()}")
    
    def _log_arbitrage_opportunity(self, cycle: Dict, rank: int):
        """Log an arbitrage opportunity."""
        logger.info(f"=== Arbitrage Opportunity #{rank} ===")
        logger.info(f"Cycle: {' -> '.join(cycle['cycle'])}")
        logger.info(f"Profit: {cycle['profit_percentage']:.4f}% ({cycle['profit']:.8f} {cycle['cycle'][0]})")
        logger.info(f"Total fees: {cycle['total_fees']:.4f}")
        
        logger.info("Steps:")
        for step in cycle['steps']:
            logger.info(f"  {step['from']} -> {step['to']} via {step['symbol']}: "
                       f"rate={step['rate']:.8f}, fee={step['fee']:.4f}, "
                       f"{step['amount_before']:.8f} -> {step['amount_after']:.8f}")
    
    def start(self):
        """Start the real-time arbitrage detection system."""
        if self.running:
            logger.warning("System already running")
            return
        
        try:
            logger.info("Starting real-time arbitrage detection system...")
            self.running = True
            
            # Start live price updates
            self.ccxt_interface.start_live_updates()
            
            # Wait for initial data
            logger.info("Waiting for initial price data...")
            initial_wait = 15
            for i in range(initial_wait):
                time.sleep(1)
                cached_prices = len(self.ccxt_interface.get_all_cached_prices())
                if i % 3 == 0:
                    logger.info(f"Received prices for {cached_prices}/{len(self.ccxt_interface.symbols)} symbols")
                
                if cached_prices >= min(20, len(self.ccxt_interface.symbols) // 3):
                    logger.info("Sufficient price data received, starting arbitrage detection")
                    break
            
            # Initial arbitrage check
            self._check_arbitrage_opportunities()
            
            logger.info("Real-time arbitrage detection system started successfully")
            logger.info("System is now monitoring for arbitrage opportunities...")
            
            # Main monitoring loop
            self._monitoring_loop()
            
        except Exception as e:
            logger.error(f"Error starting system: {e}")
            self.stop()
            raise
    
    def _monitoring_loop(self):
        """Main monitoring loop."""
        try:
            last_stats_time = 0
            stats_interval = 30  # Print stats every 30 seconds
            
            while self.running:
                time.sleep(1)
                
                current_time = time.time()
                
                # Print periodic statistics
                if current_time - last_stats_time >= stats_interval:
                    self._print_system_stats()
                    last_stats_time = current_time
                
        except KeyboardInterrupt:
            logger.info("Received keyboard interrupt signal")
            self.stop()
        except Exception as e:
            logger.error(f"Error in monitoring loop: {e}")
            self.stop()
    
    def _print_system_stats(self):
        """Print system statistics."""
        logger.info("=== System Statistics ===")
        
        # Live update stats
        update_stats = self.ccxt_interface.get_update_stats()
        logger.info(f"Price updater: {update_stats.get('running', False)}, "
                   f"fetch count: {update_stats.get('fetch_count', 0)}, "
                   f"errors: {update_stats.get('error_count', 0)}")
        
        # Price data stats
        cached_prices = len(self.ccxt_interface.get_all_cached_prices())
        logger.info(f"Price updates: {self.update_count} total, {cached_prices} symbols cached")
        
        # Show some sample prices
        sample_prices = list(self.ccxt_interface.get_all_cached_prices().items())[:3]
        for symbol, price_data in sample_prices:
            age = time.time() - price_data.get('last_updated', 0)
            logger.info(f"  {symbol}: bid={price_data.get('bid', 0):.6f}, "
                       f"ask={price_data.get('ask', 0):.6f}, age={age:.1f}s")
        
        # Cross-rate matrix stats
        matrix_stats = self.cross_rate_matrix.get_stats()
        age_seconds = matrix_stats.get('age_seconds', 0)
        age_str = f"{age_seconds:.1f}s" if age_seconds is not None else "never"
        logger.info(f"Cross-rate matrix: {matrix_stats['currencies']} currencies, "
                   f"{matrix_stats['direct_rates']} rates, "
                   f"updated {age_str} ago")
        
        # Current arbitrage opportunities
        if self.arbitrage_opportunities:
            best_profit = self.arbitrage_opportunities[0]['profit_percentage']
            logger.info(f"Arbitrage: {len(self.arbitrage_opportunities)} opportunities, "
                       f"best profit: {best_profit:.4f}%")
            
            # Show top opportunity
            top_cycle = self.arbitrage_opportunities[0]
            logger.info(f"Top cycle: {' -> '.join(top_cycle['cycle'])}")
        else:
            logger.info("Arbitrage: No opportunities found")
    
    def stop(self):
        """Stop the real-time arbitrage detection system."""
        if not self.running:
            return
        
        logger.info("Stopping real-time arbitrage detection system...")
        self.running = False
        
        # Stop live updates
        self.ccxt_interface.stop_live_updates()
        
        logger.info("Real-time arbitrage detection system stopped")
    
    def get_current_opportunities(self) -> List[Dict]:
        """Get current arbitrage opportunities."""
        return self.arbitrage_opportunities.copy()


def signal_handler(signum, frame):
    """Handle shutdown signals gracefully."""
    logger.info(f"Received signal {signum}, shutting down...")
    sys.exit(0)


def main():
    """Main function."""
    # Setup logging
    init_logger("INFO", log_file="logs/arbitrage_realtime.log", console=True)
    
    # Parse arguments
    args = config.parse_args()
    if not config.validate_args(args):
        logger.error("Invalid parameters")
        sys.exit(1)
    
    # Setup signal handlers
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    # Create and start the detector
    detector = RealTimeArbitrageDetector(
        exchange_id=args.exchange,
        limit_markets=args.limit
    )
    
    try:
        detector.start()
    except KeyboardInterrupt:
        logger.info("Keyboard interrupt received")
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
    finally:
        detector.stop()


if __name__ == "__main__":
    main()