New chat
Today
Fix Raydium SOL Pair Trading Script
7 Days
Raydium MEV Trading System Script
Enhanced Raydium MEV Trading System Script
Raydium V4 MEV Trading Implementation Guide
Get App
Fix Raydium SOL Pair Trading Script
#!/usr/bin/env python3
"""
Raydium MEV Trading System (Enhanced Version)
--------------------------------------------
High-performance MEV system for Solana/Raydium with complete support for 0x03/0x09/0x06 instructions:

1. Monitors mempool for Raydium transactions
2. Validates pool security using proven multi-layout verification
3. Executes front-running trades with comprehensive validation
4. Implements efficient balance monitoring with WebSocket subscriptions
5. Manages token accounts with robust error handling
6. Enhanced token validation for reliable trade execution
"""
import os
import sys
import json
import base64
import asyncio
import signal
import time
import random
import websockets
import logging
import traceback
import queue
import threading
import dotenv
import platform
import concurrent.futures
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Any, Optional, Set, Tuple, Union, cast
import torch
import numpy as np

# Initialize Rust bindings with comprehensive error handling
try:
    from solana_rust_bot import (
        RaydiumTradingCore, 
        TradingConfig, 
        SwapParams,
        TransactionStatus,
        PoolInfo
    )
    HAVE_RUST_BINDINGS = True
    print("Rust trading core bindings loaded successfully")
except ImportError as e:
    HAVE_RUST_BINDINGS = False
    print(f"CRITICAL: Rust trading core bindings not found: {e}")
    print("Please run: maturin develop --release")
    print(f"Current Python path: {sys.executable}")
    print(f"Current working directory: {os.getcwd()}")
    sys.exit(1)

# Load environment variables with defaults
dotenv.load_dotenv()

# Constants - Core Solana parameters
LAMPORTS_PER_SOL = 1_000_000_000
WSOL_ADDRESS = os.getenv("WSOL_ADDRESS", "So11111111111111111111111111111111111111112")
RAYDIUM_PROGRAM_ID = os.getenv("RAYDIUM_PROGRAM_ID", "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8")
TOKEN_PROGRAM_ID = os.getenv("TOKEN_PROGRAM_ID", "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
SYSTEM_PROGRAM_ID = os.getenv("SYSTEM_PROGRAM_ID", "11111111111111111111111111111111")
ATA_PROGRAM_ID = os.getenv("ATA_PROGRAM_ID", "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL")
COMPUTE_BUDGET_PROGRAM_ID = "ComputeBudget111111111111111111111111111111"
SERUM_PROGRAM_ID = os.getenv("SERUM_PROGRAM_ID", "9xQeWvG816bUx9EPjHmaT23yvVM2ZWbrrpZb9PusVFin")

# Raydium instruction types 
RAYDIUM_SWAP_INSTRUCTION_V3 = 0x03  # Instruction type for Raydium V3 swaps
RAYDIUM_SWAP_INSTRUCTION_V4 = 0x09  # Instruction type for Raydium V4 swaps
RAYDIUM_SWAP_INSTRUCTION_V6 = 0x06  # Additional supported instruction type

# Known program IDs that should never be used as token mints
# IMPORTANT: Removed whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc as it's valid for trading
KNOWN_PROGRAM_IDS = {
    RAYDIUM_PROGRAM_ID,
    TOKEN_PROGRAM_ID,
    SYSTEM_PROGRAM_ID,
    ATA_PROGRAM_ID,
    COMPUTE_BUDGET_PROGRAM_ID,
    SERUM_PROGRAM_ID,
    "LBUZKhRx4amdUADHbwgRxh3h4F6FX2rXKj5uCQVGTFT",  # Meteora DLMM Program
    "CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK",  # Raydium Concentrated Liquidity
    "9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP",  # Orca Whirlpool
    "JUP4Fb2cqiRUcaTHdrPC8h2gNsA2ETXiPDD33WcGuJB",   # Jupiter Aggregator
    "Jupiter1111111111111111111111111111111",         # Jupiter alias
    "BONK9xNjA3527soNjSuYSRY1YLRdJsD8a7zJxdYMXTFF"   # BONK token
}

# Simplified stablecoin addresses to avoid
STABLECOIN_ADDRESSES = {
    "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v",  # USDC
    "Es9vMFrzaCERmJfrF4H2FYD4KCoNkY11McCe8BenwNYB",  # USDT
}

# Transaction timing settings - ENHANCED
TRANSACTION_CONFIRMATION_TIMEOUT = 12.0  # seconds to wait for transaction confirmation
ENTRY_MIN_WAIT_TIME = 3.5  # seconds to wait after entry before checking token balance
POST_TRANSACTION_WAIT_TIME = 3.0  # seconds to wait after transaction to check results
BALANCE_CHECK_INTERVAL = 1.0  # seconds between token balance checks

# RPC rate limiting - critical to stay within 5 RPC/sec
RPC_LIMIT_PER_SECOND = 5
rpc_semaphore = asyncio.Semaphore(RPC_LIMIT_PER_SECOND)  # Max concurrent RPC calls
rpc_call_timestamps = []
rpc_call_lock = asyncio.Lock()

# Global state for inter-thread communication
opportunity_queue = queue.Queue(maxsize=100)
command_queue = queue.Queue(maxsize=10)
bot_running = threading.Event()
active_trades = set()
cached_blockhash = {"hash": None, "timestamp": 0}
balance_updates = {
    "SOL": 0.0,
    "WSOL": 0.0,
    "last_update": 0,
    "token_balances": {}  # Track all token balances
}
stats = {
    "start_time": time.time(),
    "logs_processed": 0,
    "ray_logs_found": 0,
    "instructions_parsed": 0,
    "opportunities": 0,
    "executions": 0,
    "trades_successful": 0,
    "trades_failed": 0,
    "total_profit": 0.0,
    "last_profit": None,
    "unique_pools_seen": set(),
    "rpc_errors": 0,
    "blockhash_errors": 0,
    "ata_errors": 0,
    "instruction_types": {
        "0x03": 0,
        "0x09": 0,
        "0x06": 0,
        "other": 0
    },
    "filtered_by": {
        "size_too_small": 0,
        "size_too_large": 0,
        "profit_threshold": 0,
        "jupiter": 0,
        "program_id": 0,
        "stablecoin": 0,
        "other": 0
    }
}

# Configure logging
logging.basicConfig(
    level=logging.INFO if os.getenv("LOG_LEVEL", "INFO").upper() != "DEBUG" else logging.DEBUG,
    format='%(asctime)s | %(levelname)-7s | %(message)s',
    handlers=[
        logging.FileHandler("raydium_mev.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger("raydium_mev")


class GPUAccelerator:
    """GPU acceleration for decision making and analytics using PyTorch CUDA"""
    
    def __init__(self):
        """Initialize GPU acceleration with CUDA optimization"""
        # Configure PyTorch with optimized settings from environment
        self.use_cuda = torch.cuda.is_available()
        self.device = torch.device("cuda" if self.use_cuda else "cpu")
        
        # Log GPU details
        if self.use_cuda:
            logger.info(f"GPU: {torch.cuda.get_device_name(0)}")
            logger.info(f"CUDA Version: {torch.version.cuda}")
            logger.info(f"Memory: {torch.cuda.get_device_properties(0).total_memory / 1e9:.1f}GB")
            
            # Configure CUDA optimizations from environment
            torch.backends.cuda.matmul.allow_tf32 = os.getenv("ALLOW_TF32", "true").lower() == "true"
            torch.backends.cudnn.benchmark = os.getenv("CUDA_BENCHMARK", "true").lower() == "true"
            torch.backends.cudnn.deterministic = os.getenv("CUDA_DETERMINISTIC", "false").lower() == "true"
            
            # Create CUDA stream for optimized GPU operations
            self.stream_priority = int(os.getenv("STREAM_PRIORITY", "-3"))
            self.stream = torch.cuda.Stream(priority=self.stream_priority)
            
            # Initialize timing events
            self.start_event = torch.cuda.Event(enable_timing=True, blocking=False)
            self.end_event = torch.cuda.Event(enable_timing=True, blocking=False)
            
            # Clear GPU cache
            torch.cuda.empty_cache()
            
            logger.info("GPU acceleration initialized successfully")
            
            # Initialize tensors and models
            self._init_tensors()
        else:
            logger.warning("CUDA GPU not available, using CPU only")
    
    def _init_tensors(self):
        """Initialize PyTorch tensors for trade decision acceleration"""
        try:
            # Optimize tensor size from environment
            tensor_size = int(os.getenv("TENSOR_SIZE", "176"))
            
            # Create optimized contiguous tensors
            with torch.cuda.stream(self.stream):
                self.prices = torch.randn(tensor_size, device=self.device, dtype=torch.float16).contiguous()
                self.amounts = torch.randn(tensor_size, device=self.device, dtype=torch.float16).contiguous()
                
                # Pre-compile compute function for maximum speed
                if hasattr(torch, 'compile'):
                    self._compute = torch.compile(self._compute_raw, mode="reduce-overhead", fullgraph=True)
                else:
                    self._compute = self._compute_raw
                
                # Warmup GPU operations
                for _ in range(1000):
                    self._compute(self.prices, self.amounts)
                
                # Synchronize to ensure warmup is complete
                torch.cuda.synchronize()
                
            logger.debug("GPU tensors initialized and warmed up")
        except Exception as e:
            logger.error(f"Error initializing GPU tensors: {e}")
            logger.error(traceback.format_exc())
    
    def _compute_raw(self, prices, amounts):
        """Raw compute function for optimized matrix operations"""
        return torch.matmul(prices, amounts)
    
    async def evaluate_trade(self, token_mint: str, amount_sol: float, instruction_type: int) -> Dict[str, float]:
        """Evaluate a potential trade with GPU acceleration
        
        Args:
            token_mint: Token mint address
            amount_sol: Trade amount in SOL
            instruction_type: Instruction type (0x03, 0x09, 0x06)
            
        Returns:
            Dict with estimated profit and execution probability
        """
        if not self.use_cuda:
            # Fallback to CPU calculation
            return {
                "estimated_profit": amount_sol * 0.02 - 0.00095*2,  # Higher profit rate for 0x03
                "execution_probability": 0.90
            }
        
        try:
            # Create token identifier without trying to convert to int
            # Just use sum of character codes as a simple hash
            token_chars = token_mint[:12] if len(token_mint) >= 12 else token_mint
            token_id = sum(ord(c) for c in token_chars) % 1000
            
            # Use fast GPU compute for evaluation
            # Fix: Use the new PyTorch amp.autocast API
            with torch.amp.autocast(device_type='cuda', enabled=True):
                with torch.cuda.stream(self.stream):
                    self.start_event.record()
                    
                    # Convert inputs to tensors
                    amount_tensor = torch.tensor([amount_sol], device=self.device, dtype=torch.float16)
                    token_id_tensor = torch.tensor([token_id], device=self.device, dtype=torch.float16)
                    instr_tensor = torch.tensor([instruction_type], device=self.device, dtype=torch.float16)
                    
                    # Simple profit model with different rates based on instruction type
                    profit_rate = 0.02  # Base rate 2.0%
                    
                    # Higher profit rate for V4 (0x09) instructions
                    if instruction_type == RAYDIUM_SWAP_INSTRUCTION_V4:
                        profit_rate = 0.025  # 2.5% base rate for V4
                    
                    # Adjust for larger amounts
                    if amount_sol > 40:
                        profit_rate += 0.01  # Add 1% for large amounts
                    elif amount_sol > 20:
                        profit_rate += 0.005  # Add 0.5% for medium amounts
                    
                    # Calculate estimated profit
                    gross_profit = amount_sol * profit_rate
                    tx_fee = 0.00095  # Fixed fee per transaction
                    net_profit = gross_profit - (tx_fee * 2)  # Entry + exit fees
                    
                    self.end_event.record()
                    self.end_event.synchronize()
                    
                    # Get execution time
                    execution_time = self.start_event.elapsed_time(self.end_event) / 1000.0
                    logger.debug(f"GPU trade evaluation completed in {execution_time:.6f}s")
            
            return {
                "estimated_profit": max(0, net_profit),
                "execution_probability": 0.90 if instruction_type == RAYDIUM_SWAP_INSTRUCTION_V3 else 0.92
            }
                
        except Exception as e:
            logger.error(f"Error in GPU trade evaluation: {e}")
            # Fallback to CPU calculation
            return {
                "estimated_profit": amount_sol * 0.02 - 0.00095*2,
                "execution_probability": 0.85
            }
    
    def cleanup(self):
        """Clean up GPU resources"""
        if self.use_cuda:
            torch.cuda.empty_cache()


class RaydiumMEVBot:
    """High-performance Raydium MEV bot with robust transaction execution"""
    
    def __init__(self):
        """Initialize the MEV bot with proper configuration validation"""
        logger.info("Initializing Raydium MEV Bot...")
        
        # Load configuration from environment with validation
        self.config = self.load_env_config()
        
        # Initialize GPU accelerator for FP16 precision on 4070 Ti
        self.gpu = GPUAccelerator()
        
        # Trading parameters with proper validation
        self.front_run_amount = float(self.config.get("FRONT_RUN_AMOUNT", 0.3))
        self.min_victim_size = float(self.config.get("SANDWICH_MIN_VICTIM_SIZE", 5.0))
        self.max_victim_size = float(self.config.get("SANDWICH_MAX_VICTIM_SIZE", 320.0))
        self.min_profit = float(self.config.get("MIN_PROFIT", 0.002))
        self.min_trade_interval = float(self.config.get("MIN_TRADE_INTERVAL", 0.05))
        self.slippage_bps = int(self.config.get("SLIPPAGE_BPS", 1000))
        self.exit_slippage_bps = int(self.config.get("SLIPPAGE_BPS", 1000)) + 500  # Add 5% for exits
        self.fixed_gas = 0.00095  # Fixed gas cost estimate per transaction
        
        # Security validation parameters
        self.require_mint_revoked = self.config.get("VERIFY_MINT_AUTHORITY", "true").lower() == "true"
        self.min_liquidity_lock = float(self.config.get("MIN_LIQUIDITY_LOCK_PERCENT", 95.0))
        self.min_exit_liquidity = float(self.config.get("MIN_EXIT_LIQUIDITY_SOL", 5.0))
        
        # RPC parameters
        self.rpc_retry_attempts = int(self.config.get("MAX_RETRIES", 2))
        self.max_exit_retries = int(self.config.get("MAX_EXIT_RETRIES", 3))
        self.rpc_retry_delay = 0.5
        
        # Create required directories
        os.makedirs("logs", exist_ok=True)
        os.makedirs("config", exist_ok=True)
        
        # Initialize trading core (Rust)
        self.initialize_trading_core()
        
        # State tracking
        self.last_execution_time = 0
        self.seen_txs = set()
        self.pool_security_cache = {}
        self.token_resolution_cache = {}
        self.ata_cache = {}  # Cache for token ATAs
        self.current_position = None
        self.target_pools = {}
        
        # WebSocket balance monitoring
        self.wsol_ata = self.config.get("WSOL_TOKEN_ACCOUNT")
        if not self.wsol_ata:
            logger.error("WSOL_TOKEN_ACCOUNT not found in environment! Please set this value.")
        
        # Global trade lock for thread safety
        self.trade_lock = asyncio.Lock()
        
        logger.info(f"Bot initialized with {self.front_run_amount} SOL front-running amount")
        logger.info(f"Target victim size: {self.min_victim_size} - {self.max_victim_size} SOL")
        logger.info(f"Minimum profit: {self.min_profit} SOL")
        
        # Log RPC config
        logger.info(f"Using RPC URL: {self.config.get('RPC_URL')}")
        logger.info(f"Using WebSocket URL: {self.config.get('WS_URL')}")
    
    def load_env_config(self) -> Dict[str, str]:
        """Load and validate configuration from environment variables"""
        # Load .env file with error checking
        try:
            dotenv.load_dotenv()
            logger.info("Environment configuration loaded from .env file")
        except Exception as e:
            logger.warning(f"Failed to load .env file: {e}")
        
        # Build configuration dictionary with validation
        config = {}
        
        # Critical configuration parameters with validation
        critical_params = {
            "RPC_URL": "https://api.mainnet-beta.solana.com",
            "WS_URL": "wss://api.mainnet-beta.solana.com",
            "KEYPAIR_PATH": "keypair.bin",
            "COMPUTE_LIMIT": "50000",
            "PRIORITY_FEE": "950000",
            "FRONT_RUN_AMOUNT": "0.3"
        }
        
        # Validate all critical parameters
        for key, default in critical_params.items():
            value = os.environ.get(key, default)
            config[key] = value
            
            # Log if using default for critical parameter
            if key in os.environ:
                logger.info(f"Using configured {key}: {value if 'KEY' not in key else '[REDACTED]'}")
            else:
                logger.warning(f"Using default {key}: {value if 'KEY' not in key else '[REDACTED]'}")
        
        # Add remaining environment variables
        for key, value in os.environ.items():
            if key not in config:
                config[key] = value
        
        # Validate numeric parameters
        numeric_params = [
            "FRONT_RUN_AMOUNT", "SANDWICH_MIN_VICTIM_SIZE", "SANDWICH_MAX_VICTIM_SIZE",
            "MIN_PROFIT", "SLIPPAGE_BPS", "COMPUTE_LIMIT", "PRIORITY_FEE"
        ]
        
        for param in numeric_params:
            if param in config:
                try:
                    float(config[param])  # Validate numeric
                except ValueError:
                    logger.error(f"Invalid numeric value for {param}: {config[param]}")
                    # Set to default if invalid
                    if param == "FRONT_RUN_AMOUNT":
                        config[param] = "0.3"
                    elif param == "SANDWICH_MIN_VICTIM_SIZE":
                        config[param] = "5.0"
                    elif param == "SANDWICH_MAX_VICTIM_SIZE":
                        config[param] = "320.0"
                    elif param == "MIN_PROFIT":
                        config[param] = "0.002"
                    elif param == "SLIPPAGE_BPS":
                        config[param] = "1000"
                    elif param == "COMPUTE_LIMIT":
                        config[param] = "50000"
                    elif param == "PRIORITY_FEE":
                        config[param] = "950000"
        
        return config
            
    def initialize_trading_core(self):
        """Initialize the Rust trading core with proper error handling"""
        try:
            # Validate keypair path
            keypair_path = self.config.get("KEYPAIR_PATH", "keypair.bin")
            if not os.path.exists(keypair_path):
                logger.critical(f"Keypair file not found: {keypair_path}")
                raise FileNotFoundError(f"Keypair file not found: {keypair_path}")
                
            # Configure trading parameters with type checking
            compute_units = int(self.config.get("COMPUTE_LIMIT", "50000"))
            priority_fee = int(self.config.get("PRIORITY_FEE", "950000"))
            slippage_bps = self.slippage_bps
            max_retries = self.rpc_retry_attempts
            commitment = self.config.get("COMMITMENT_LEVEL", "processed")
            skip_preflight = self.config.get("SKIP_PREFLIGHT", "true").lower() == "true"
            
            # Validate RPC URL format
            rpc_url = self.config.get("RPC_URL", "https://winny-rychu7-fast-mainnet.helius-rpc.com")
            if not rpc_url.startswith(("http://", "https://")):
                logger.warning(f"Invalid RPC URL format: {rpc_url}, using default")
                rpc_url = "https://winny-rychu7-fast-mainnet.helius-rpc.com"
                
            # Create trading config
            trading_config = TradingConfig(
                compute_units=compute_units,
                priority_fee=priority_fee,
                slippage_bps=slippage_bps,
                max_retries=max_retries,
                commitment=commitment,
                skip_preflight=skip_preflight
            )
            
            # Create the RaydiumTradingCore instance
            self.trading_core = RaydiumTradingCore(
                rpc_url=rpc_url,
                keypair_path=keypair_path,
                config=trading_config
            )
            
            # Debug: Print available methods to identify correct names
            logger.debug("Available RaydiumTradingCore methods:")
            methods = [m for m in dir(self.trading_core) if not m.startswith('_')]
            logger.debug(str(methods))
            
            # Get public key for verification - try different accessor methods
            try:
                # First try to access as an attribute/property
                if hasattr(self.trading_core, "wallet_address"):
                    self.public_key = self.trading_core.wallet_address
                # Then try common getter methods
                elif hasattr(self.trading_core, "get_wallet_address"):
                    self.public_key = self.trading_core.get_wallet_address()
                else:
                    # Use the first method that has "public" or "wallet" in the name
                    key_methods = [m for m in methods if "public" in m.lower() or "wallet" in m.lower()]
                    if key_methods:
                        self.public_key = getattr(self.trading_core, key_methods[0])()
                    else:
                        # Last resort - since we know the public key should be in env vars
                        self.public_key = self.config.get("SOLANA_PUBLIC_KEY", "Unknown")
                        logger.warning(f"Could not determine public key from trading core, using config value")
                
                logger.info(f"Trading core initialized with public key: {self.public_key}")
                
                # Verify public key matches config if provided
                expected_public_key = self.config.get("SOLANA_PUBLIC_KEY")
                if expected_public_key and expected_public_key != self.public_key:
                    logger.warning(f"Public key mismatch: {self.public_key} (actual) vs {expected_public_key} (config)")
                    
            except Exception as e:
                logger.error(f"Error getting public key: {e}")
                # Fallback to config value
                self.public_key = self.config.get("SOLANA_PUBLIC_KEY", "Unknown")
                logger.warning(f"Using public key from config: {self.public_key}")
            
        except Exception as e:
            logger.critical(f"Failed to initialize trading core: {e}")
            logger.critical(traceback.format_exc())
            raise RuntimeError(f"Trading core initialization failed: {e}")

    async def run(self):
        """Run the MEV bot with proper signal handling and cleanup"""
        logger.info("Starting Raydium MEV Bot...")
        bot_running.set()
        
        try:
            # Start worker threads
            self.start_worker_threads()
            
            # Start balance monitoring
            balance_task = asyncio.create_task(self.monitor_balances())
            
            # Get initial balances
            await self.check_balances()
            
            # Start monitoring WebSocket
            await self.monitor_mempool()
            
        except KeyboardInterrupt:
            logger.info("Keyboard interrupt detected, shutting down...")
        except Exception as e:
            logger.error(f"Error in main loop: {e}")
            logger.error(traceback.format_exc())
        finally:
            # Signal threads to stop
            bot_running.clear()
            command_queue.put(("shutdown", None))
            
            # Cancel balance monitoring
            if 'balance_task' in locals():
                balance_task.cancel()
                try:
                    await balance_task
                except asyncio.CancelledError:
                    pass
            
            # Ensure we exit any remaining positions
            if self.current_position is not None:
                logger.info("Exiting remaining position before shutdown")
                await self.exit_position(self.current_position)
            
            # Clean up GPU resources
            self.gpu.cleanup()
            
            logger.info("Bot shutdown complete")
    
    def start_worker_threads(self):
        """Start worker threads for trade execution and monitoring"""
        # Start execution thread
        self.execution_thread = threading.Thread(
            target=self.execution_worker, 
            name="execution_worker", 
            daemon=True
        )
        self.execution_thread.start()
        
        # Start stats thread
        self.stats_thread = threading.Thread(
            target=self.stats_worker, 
            name="stats_worker", 
            daemon=True
        )
        self.stats_thread.start()
        
        logger.info("Worker threads started")
    
    async def rate_limited_call_with_backoff(self, coro):
        """Execute coroutine with exponential backoff for rate limit errors"""
        retry_count = 0
        max_retries = self.rpc_retry_attempts + 2  # Add extra retries for rate limit errors
        base_delay = 0.5
        
        while retry_count <= max_retries:
            try:
                return await self.rate_limited_call(coro)
            except Exception as e:
                error_str = str(e).lower()
                retry_count += 1
                
                # Check if it's a rate limit error
                is_rate_limit = "rate" in error_str and "limit" in error_str
                
                if retry_count <= max_retries:
                    # Calculate delay with exponential backoff and jitter
                    delay = base_delay * (2 ** retry_count)
                    if is_rate_limit:
                        delay *= 1.5  # Extra delay for rate limit errors
                    
                    jitter = random.uniform(0, 0.1 * delay)
                    wait_time = delay + jitter
                    
                    logger.warning(f"RPC error (attempt {retry_count}/{max_retries}), " +
                                   f"retrying in {wait_time:.2f}s: {e}")
                    await asyncio.sleep(wait_time)
                else:
                    raise
    
    async def rate_limited_call(self, coro):
        """Execute coroutine with rate limiting and exponential backoff retry"""
        # Track RPC calls to stay within limits (5 RPC/sec)
        async with rpc_call_lock:
            current_time = time.time()
            # Remove timestamps older than 1 second
            global rpc_call_timestamps
            rpc_call_timestamps = [t for t in rpc_call_timestamps if current_time - t < 1.0]
           
            # Check if we're at the RPC limit
            if len(rpc_call_timestamps) >= RPC_LIMIT_PER_SECOND:
                # Calculate time to wait - add a small buffer (50ms) to be safe
                wait_time = 1.0 - (current_time - rpc_call_timestamps[0]) + 0.05
                if wait_time > 0:
                    logger.debug(f"RPC rate limit reached, waiting {wait_time:.3f}s")
                    await asyncio.sleep(wait_time)
                   
                    # After waiting, clear old entries again
                    current_time = time.time()
                    rpc_call_timestamps = [t for t in rpc_call_timestamps if current_time - t < 1.0]
           
            # Add current timestamp to the list before executing
            rpc_call_timestamps.append(time.time())
       
        # Execute with semaphore for concurrent control
        async with rpc_semaphore:
            retry_count = 0
            max_retries = self.rpc_retry_attempts
            last_error = None
            
            while retry_count <= max_retries:
                try:
                    result = await coro
                    return result
                except Exception as e:
                    retry_count += 1
                    last_error = e
                    
                    # Log the error with more details
                    logger.warning(f"RPC call error (attempt {retry_count}/{max_retries+1}): {str(e)}")
                    
                    # Check if we should retry
                    if retry_count <= max_retries:
                        # If it's a rate limit error, wait longer
                        if "429" in str(e) or "rate limit" in str(e).lower():
                            stats["rpc_errors"] += 1
                            # Exponential backoff with jitter
                            base_wait = self.rpc_retry_delay * (2 ** retry_count)
                            jitter = random.uniform(0, 0.1 * base_wait)  # 10% jitter
                            wait_time = base_wait + jitter
                            logger.warning(f"Rate limit hit, retrying in {wait_time:.2f}s")
                            await asyncio.sleep(wait_time)
                            
                            # Remove old timestamps after waiting
                            async with rpc_call_lock:
                                current_time = time.time()
                                rpc_call_timestamps = [t for t in rpc_call_timestamps if current_time - t < 1.0]
                        else:
                            # For other errors, shorter retry
                            wait_time = self.rpc_retry_delay * retry_count
                            logger.warning(f"Retrying in {wait_time:.2f}s")
                            await asyncio.sleep(wait_time)
                    else:
                        # Give up after max retries
                        logger.error(f"Failed after {max_retries+1} attempts: {last_error}")
                        raise last_error
    
    async def check_balances(self):
        """Check SOL and token balances with proper error handling"""
        try:
            # Get SOL balance
            sol_balance = await self.rate_limited_call(self.trading_core.get_sol_balance())
            
            # Log SOL balance
            logger.info(f"SOL balance: {sol_balance:.6f} SOL")
            
            # Update global balance tracking
            balance_updates["SOL"] = sol_balance
            balance_updates["last_update"] = time.time()
            
            # Check WSOL balance
            wsol_balance = await self.rate_limited_call(
                self.trading_core.get_token_balance_py(WSOL_ADDRESS)
            )
            logger.info(f"WSOL balance: {wsol_balance:.6f} SOL")
            balance_updates["WSOL"] = wsol_balance
            
            # Update token balances cache
            balance_updates["token_balances"] = {WSOL_ADDRESS: wsol_balance}
            
            # Check minimum balance requirements
            min_sol_balance = float(self.config.get("MIN_SOL_BALANCE", 0.03))
            min_wsol_balance = float(self.config.get("MIN_WSOL_BALANCE", 0.02))
            
            if sol_balance < min_sol_balance:
                logger.warning(f"SOL balance ({sol_balance:.6f}) below minimum ({min_sol_balance:.6f})")
            
            if wsol_balance < min_wsol_balance:
                logger.warning(f"WSOL balance ({wsol_balance:.6f}) below minimum ({min_wsol_balance:.6f})")
            
            # Wrap SOL to WSOL if needed and ratio is off
            target_wsol_ratio = float(self.config.get("WSOL_TARGET_RATIO", 0.95))
            if wsol_balance < self.front_run_amount * 1.5 and sol_balance > min_sol_balance + self.front_run_amount:
                logger.info(f"Wrapping SOL to WSOL to reach target ratio {target_wsol_ratio}")
                # The Rust core will handle the actual wrapping when a trade is executed
            
        except Exception as e:
            logger.error(f"Failed to check balances: {e}")
            logger.error(traceback.format_exc())
    
    async def monitor_balances(self):
        """Monitor SOL and WSOL balances via WebSocket subscription using correct Helius format"""
        retry_delay = 1
        max_retry_delay = 30
        
        # Use the proper WebSocket URL from config
        ws_url = self.config.get("WS_URL")
        api_key = self.config.get("API_KEY")
        
        # Prepare headers with API key if provided
        headers = {}
        if api_key:
            headers["api-key"] = api_key
        
        logger.info(f"Starting balance monitoring WebSocket")
        
        while bot_running.is_set():
            try:
                # Connect with improved connection parameters
                async with websockets.connect(
                    ws_url,
                    extra_headers=headers,
                    ping_interval=15,
                    ping_timeout=60,
                    close_timeout=120
                ) as ws:
                    logger.info("Connected to balance monitoring WebSocket")
                    
                    # Accounts to monitor - use hardcoded addresses for stability
                    accounts = {
                        "SOL": self.public_key, 
                        "WSOL": self.wsol_ata
                    }
                    
                    # Subscribe to each account with fixed ID (1)
                    subscriptions = {}
                    id_counter = 1
                    
                    for acc_type, address in accounts.items():
                        if not address:
                            continue
                        
                        logger.info(f"Subscribing to {acc_type} account: {address}")
                        
                        # Use fixed format for subscription
                        await ws.send(
                            json.dumps({
                                "jsonrpc": "2.0",
                                "id": id_counter,
                                "method": "accountSubscribe",
                                "params": [
                                    address,
                                    {"encoding": "base64", "commitment": "confirmed"}
                                ]
                            })
                        )
                        
                        # Store ID to subscription mapping
                        subscriptions[id_counter] = acc_type
                        id_counter += 1
                    
                    # Reset retry delay on successful connection
                    retry_delay = 1
                    
                    # Process WebSocket messages
                    while bot_running.is_set():
                        try:
                            msg = await asyncio.wait_for(ws.recv(), timeout=30)
                            data = json.loads(msg)
                            
                            # Store subscription IDs
                            if "result" in data and "id" in data:
                                sub_id = data["id"]
                                acc_type = subscriptions.get(sub_id, "Unknown")
                                if acc_type in ["SOL", "WSOL"]:
                                    subscriptions[data["result"]] = acc_type
                                    logger.info(f"{acc_type} subscription confirmed: {data['result']}")
                                continue
                            
                            # Handle account notifications
                            if "method" in data and data["method"] == "accountNotification":
                                sub_id = data["params"]["subscription"]
                                acc_type = subscriptions.get(sub_id, "Unknown")
                                
                                if acc_type:
                                    # Extract lamports for SOL account
                                    if acc_type == "SOL":
                                        lamports = data["params"]["result"]["value"]["lamports"]
                                        balance = lamports / LAMPORTS_PER_SOL
                                        balance_updates["SOL"] = balance
                                        balance_updates["last_update"] = time.time()
                                        logger.info(f"SOL Balance updated: {balance:.6f} SOL")
                                    
                                    # Parse token account data for WSOL
                                    elif acc_type == "WSOL":
                                        # Decode base64 data from the account
                                        encoded_data = data["params"]["result"]["value"]["data"][0]
                                        decoded_data = base64.b64decode(encoded_data)
                                        
                                        # SPL Token account layout - amount at offset 64
                                        if len(decoded_data) >= 72:
                                            amount = int.from_bytes(decoded_data[64:72], byteorder="little")
                                            wsol_balance = amount / LAMPORTS_PER_SOL
                                            balance_updates["WSOL"] = wsol_balance
                                            balance_updates["last_update"] = time.time()
                                            logger.info(f"WSOL Balance updated: {wsol_balance:.6f} SOL")
                                            
                                            # Store in token balances
                                            balance_updates["token_balances"][WSOL_ADDRESS] = wsol_balance
                        
                        except asyncio.TimeoutError:
                            # Send ping to keep connection alive
                            pong_waiter = await ws.ping()
                            try:
                                await asyncio.wait_for(pong_waiter, timeout=10)
                            except asyncio.TimeoutError:
                                # No response to ping, connection may be dead
                                logger.warning("No pong response, reconnecting...")
                                break
                        except websockets.ConnectionClosed:
                            logger.warning("Balance monitoring connection closed")
                            break
                        
            except Exception as e:
                logger.error(f"Balance monitoring error: {e}")
                
                if not bot_running.is_set():
                    break
                    
                # Retry with backoff
                logger.info(f"Reconnecting balance monitor in {retry_delay}s...")
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, max_retry_delay)
    
    async def monitor_mempool(self):
        """Monitor mempool for Raydium transactions with proper error handling"""
        retry_delay = 1
        max_retry_delay = 30
        
        # Use Helius WebSocket URL with explicit API key
        ws_url = self.config.get("HELIUS_WS_URL") or self.config.get("WS_URL")
        api_key = self.config.get("API_KEY")
        
        # Prepare headers
        headers = {}
        if api_key:
            headers["api-key"] = api_key
        
        logger.info(f"Monitoring mempool for Raydium V3/V4/V6 (0x03/0x09/0x06) transactions")
        
        while bot_running.is_set():
            try:
                # Connect to WebSocket with improved parameters
                async with websockets.connect(
                    ws_url,
                    extra_headers=headers,
                    ping_interval=15,
                    ping_timeout=60,
                    close_timeout=120
                ) as ws:
                    logger.info("Connected to mempool WebSocket")
                    
                    # Subscribe to Raydium program logs
                    subscription_msg = json.dumps({
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "logsSubscribe",
                        "params": [
                            {"mentions": [RAYDIUM_PROGRAM_ID]},
                            {"commitment": self.config.get("COMMITMENT_LEVEL", "processed")}
                        ]
                    })
                    
                    await ws.send(subscription_msg)
                    
                    # Check subscription confirmation
                    response = await ws.recv()
                    confirmation = json.loads(response)
                    
                    if "result" in confirmation:
                        subscription_id = confirmation["result"]
                        logger.info(f"Subscription successful (ID: {subscription_id})")
                    else:
                        logger.error(f"Subscription failed: {confirmation}")
                        raise Exception("Failed to subscribe to logs")
                    
                    # Reset retry delay on successful connection
                    retry_delay = 1
                    
                    # Process WebSocket messages
                    while bot_running.is_set():
                        try:
                            # Receive message with timeout
                            msg = await asyncio.wait_for(ws.recv(), timeout=30)
                            
                            # Process the message
                            await self.process_mempool_message(msg)
                            
                        except asyncio.TimeoutError:
                            # Send ping to keep connection alive
                            pong_waiter = await ws.ping()
                            try:
                                await asyncio.wait_for(pong_waiter, timeout=10)
                            except asyncio.TimeoutError:
                                # No response to ping, connection may be dead
                                logger.warning("No pong response, reconnecting...")
                                break
                        except websockets.ConnectionClosed:
                            logger.warning("WebSocket connection closed")
                            break
            
            except Exception as e:
                logger.error(f"WebSocket error: {e}")
                
                # Check if bot is still running
                if not bot_running.is_set():
                    break
                
                # Exponential backoff for reconnection
                logger.warning(f"Reconnecting in {retry_delay} seconds...")
                await asyncio.sleep(retry_delay)
                retry_delay = min(retry_delay * 2, max_retry_delay)
    
    async def process_mempool_message(self, message: str):
        """Parse and process a mempool message with rigorous validation"""
        try:
            # Parse the message
            data = json.loads(message)
            
            # Update stats
            stats["logs_processed"] += 1
            
            # Check if valid notification with logs
            if "params" in data and "result" in data["params"]:
                result = data["params"]["result"]
                
                # Extract logs and signature
                if "value" in result:
                    value = result["value"]
                    logs = value.get("logs", [])
                    signature = value.get("signature", "unknown")
                    
                    # Skip if no logs or already seen
                    if not logs or signature in self.seen_txs:
                        return
                    
                    # Add to seen transactions
                    self.seen_txs.add(signature)
                    if len(self.seen_txs) > 5000:  # Limit cache size
                        self.seen_txs = set(list(self.seen_txs)[-2500:])
                    
                    # Skip Jupiter transactions early for efficiency
                    if any("Jupiter" in log for log in logs):
                        stats["filtered_by"]["jupiter"] += 1
                        return
                    
                    # Check if in a position and need to exit
                    if self.current_position is not None:
                        command_queue.put(("check_exit", signature))
                        return
                    
                    # Check for ray_log entries
                    for log in logs:
                        if "ray_log:" in log:
                            stats["ray_logs_found"] += 1
                            
                            # Decode and process the ray_log
                            decoded = self.decode_ray_log(log)
                            if decoded:
                                # Process supported Raydium instructions (0x03, 0x09, 0x06)
                                instr_type = decoded.get("instruction_type", -1)
                                
                                # Check if this is a supported instruction type
                                if instr_type == RAYDIUM_SWAP_INSTRUCTION_V3:
                                    stats["instructions_parsed"] += 1
                                    stats["instruction_types"]["0x03"] += 1
                                elif instr_type == RAYDIUM_SWAP_INSTRUCTION_V4:
                                    stats["instructions_parsed"] += 1
                                    stats["instruction_types"]["0x09"] += 1
                                elif instr_type == RAYDIUM_SWAP_INSTRUCTION_V6:
                                    stats["instructions_parsed"] += 1
                                    stats["instruction_types"]["0x06"] += 1
                                else:
                                    # Track other types in stats
                                    stats["instruction_types"]["other"] += 1
                                    continue
                                
                                # Add signature to decoded data for tracking
                                decoded["signature"] = signature
                                
                                # Process the decoded instruction
                                await self.process_instruction(decoded, signature, logs)
        
        except json.JSONDecodeError:
            # Ignore invalid JSON
            pass
        except Exception as e:
            logger.error(f"Error processing mempool message: {e}")
            logger.error(traceback.format_exc())
    
    def decode_ray_log(self, log_line: str) -> Optional[Dict]:
        """Decode Raydium ray_log data from log line with support for V3/V4/V6 instructions"""
        try:
            if "ray_log:" not in log_line:
                return None
            
            # Extract the base64 encoded part after ray_log:
            ray_log_part = log_line.split("ray_log:")[1].strip()
            raw_data = base64.b64decode(ray_log_part)
            
            if len(raw_data) < 9:  # Need at least 9 bytes for valid instruction
                return None
            
            # Parse instruction type (first byte)
            instr_type = raw_data[0]
            
            # Log all instruction types for debugging
            logger.debug(f"Found instruction type: 0x{instr_type:02x}")
            
            # Important fix: For the Raydium logs, the instruction type might be encoded differently
            # Sometimes the type is stored as an int value (9) rather than hex (0x09)
            # Let's handle this conversion correctly
            if instr_type == 9:
                instr_type = RAYDIUM_SWAP_INSTRUCTION_V4  # Convert to our constant
            
            # Check if this is one of our supported instruction types
            if instr_type not in [RAYDIUM_SWAP_INSTRUCTION_V3, RAYDIUM_SWAP_INSTRUCTION_V4, RAYDIUM_SWAP_INSTRUCTION_V6]:
                return None
            
            # Special case for V4 (0x09) instruction type
            if instr_type == RAYDIUM_SWAP_INSTRUCTION_V4:
                # V4 uses a discriminator in the first byte, but the amount is still at bytes 1-9
                amount_bytes = raw_data[1:9]
                amount_lamports = int.from_bytes(amount_bytes, byteorder="little")
                amount_sol = amount_lamports / LAMPORTS_PER_SOL
                
                # Extract additional values if available
                value_2 = 0
                value_3 = 0
                value_4 = 0
                
                if len(raw_data) >= 17:
                    value_2 = int.from_bytes(raw_data[9:17], byteorder="little")
                if len(raw_data) >= 25:
                    value_3 = int.from_bytes(raw_data[17:25], byteorder="little")
                if len(raw_data) >= 33:
                    value_4 = int.from_bytes(raw_data[25:33], byteorder="little")
                
                logger.debug(f"Decoded Raydium V4 instruction: 0x{instr_type:02x}, amount: {amount_sol:.6f} SOL")
                
                return {
                    "instruction_type": RAYDIUM_SWAP_INSTRUCTION_V4,  # Explicitly set to V4 constant
                    "amount_lamports": amount_lamports,
                    "amount_sol": amount_sol,
                    "value_2": value_2,
                    "value_2_sol": value_2 / LAMPORTS_PER_SOL,
                    "value_3": value_3,
                    "value_3_sol": value_3 / LAMPORTS_PER_SOL,
                    "value_4": value_4,
                    "value_4_sol": value_4 / LAMPORTS_PER_SOL,
                    "direction": "buy",  # Default direction
                    "timestamp": datetime.now().strftime('%H:%M:%S.%f')[:-3],
                    "raw_data": raw_data
                }
            else:
                # Standard processing for V3 (0x03) or V6 (0x06) instructions
                amount_bytes = raw_data[1:9]
                amount_lamports = int.from_bytes(amount_bytes, byteorder="little")
                amount_sol = amount_lamports / LAMPORTS_PER_SOL
                
                # For 0x03 instruction, extract additional data (in 8-byte chunks)
                value_2 = 0 
                value_3 = 0
                value_4 = 0
                
                # Extract additional values if available
                if len(raw_data) >= 17:
                    value_2 = int.from_bytes(raw_data[9:17], byteorder="little")
                if len(raw_data) >= 25:
                    value_3 = int.from_bytes(raw_data[17:25], byteorder="little")
                if len(raw_data) >= 33:
                    value_4 = int.from_bytes(raw_data[25:33], byteorder="little")
                
                # Log the decoded instruction
                logger.debug(f"Decoded Raydium V{3 if instr_type == RAYDIUM_SWAP_INSTRUCTION_V3 else 6} instruction: 0x{instr_type:02x}, amount: {amount_sol:.6f} SOL")
                
                return {
                    "instruction_type": instr_type,
                    "amount_lamports": amount_lamports,
                    "amount_sol": amount_sol,
                    "value_2": value_2,
                    "value_2_sol": value_2 / LAMPORTS_PER_SOL,
                    "value_3": value_3,
                    "value_3_sol": value_3 / LAMPORTS_PER_SOL,
                    "value_4": value_4,
                    "value_4_sol": value_4 / LAMPORTS_PER_SOL,
                    "direction": "buy",  # Default for v3 instructions
                    "timestamp": datetime.now().strftime('%H:%M:%S.%f')[:-3],
                    "raw_data": raw_data
                }
        except Exception as e:
            logger.error(f"Error decoding ray_log: {e}")
            logger.error(traceback.format_exc())
            return None

    def extract_token_info(self, logs: List[str]) -> Optional[Tuple[str, str]]:
        """Extract token mint and pool ID from transaction logs - simplified for better reliability"""
        try:
            # Known program IDs to exclude
            known_programs = KNOWN_PROGRAM_IDS.copy()  # Make a copy to avoid modifying the original
            
            # Pull out all potential Solana addresses from logs
            token_mint = None
            pool_id = None
            potential_addresses = []
            potential_tokens = []
            potential_pools = []
            
            # Extract program invocations and token accounts
            for log in logs:
                # Look for program invocations
                if "Program " in log and " invoke" in log:
                    parts = log.split()
                    if len(parts) >= 2:
                        program_id = parts[1]
                        # Skip if program is known
                        if program_id not in known_programs and len(program_id) >= 32:
                            potential_addresses.append(program_id)
                            
                # Look for SPL Token references
                if "spl-token" in log.lower():
                    words = log.split()
                    for word in words:
                        if len(word) >= 32 and word.isalnum() and word not in known_programs:
                            potential_tokens.append(word)
                
                # Extract any account addresses from the logs  
                words = log.split()
                for word in words:
                    if (len(word) >= 32 and len(word) <= 44 and 
                        word.isalnum() and 
                        word not in known_programs and
                        word not in potential_addresses):
                        potential_addresses.append(word)
            
            # First, look for explicit token mint references
            for log in logs:
                if "mint" in log.lower() or "token mint" in log.lower():
                    for addr in potential_addresses:
                        if addr in log and token_mint is None:
                            token_mint = addr
                            potential_tokens.append(addr)
                            
                # Look for pool mentions
                if "pool" in log.lower() or "liquidity" in log.lower() or "amm" in log.lower():
                    for addr in potential_addresses:
                        if addr in log and pool_id is None:
                            pool_id = addr
                            potential_pools.append(addr)
            
            # Special case: If we see the LBUZKhRx prefix, that's likely the Meteora program
            meteora_program = "LBUZKhRx4amdUADHbwgRxh3h4F6FX2rXKj5uCQVGTFT"
            for addr in potential_addresses:
                if addr.startswith("LBUZKhRx"):
                    # This is likely the Meteora program - don't use as token
                    if addr not in known_programs:
                        known_programs.add(addr)
                    if token_mint == addr:
                        token_mint = None  # Reset if we mistakenly assigned it
            
            # If we don't have both, try the potential tokens and potential pools
            if token_mint is None and len(potential_tokens) > 0:
                for token in potential_tokens:
                    if token not in known_programs and not token.startswith("LBUZKhRx"):
                        token_mint = token
                        break
            
            if pool_id is None and len(potential_pools) > 0:
                pool_id = potential_pools[0]
            
            # If we still don't have both, use first two unique addresses
            if (pool_id is None or token_mint is None) and len(potential_addresses) >= 2:
                unique_addrs = []
                for addr in potential_addresses:
                    if addr not in unique_addrs and addr not in known_programs and not addr.startswith("LBUZKhRx"):
                        unique_addrs.append(addr)
                
                if len(unique_addrs) >= 2:
                    if pool_id is None:
                        pool_id = unique_addrs[0]
                    if token_mint is None:
                        token_mint = unique_addrs[1]
            
            # If we have both, verify token mint isn't a known program ID and isn't Meteora
            if pool_id and token_mint:
                # Verify token_mint isn't a program ID or the same as pool_id
                if token_mint in known_programs or token_mint == pool_id:
                    logger.debug(f"Ignoring program ID as token mint: {token_mint[:8]}...")
                    return None
                
                # Double check both are valid-looking addresses
                if len(token_mint) < 32 or len(pool_id) < 32:
                    logger.debug(f"Invalid address length detected: {token_mint[:8]}... or {pool_id[:8]}...")
                    return None
                
                # Check for the Meteora program prefix
                if token_mint.startswith("LBUZKhRx"):
                    logger.debug(f"Ignoring LBUZKhRx program as token: {token_mint[:12]}...")
                    return None
                    
                logger.debug(f"Extracted token info: {token_mint[:8]}... (pool: {pool_id[:8]}...)")
                
                # Track unique pools seen
                stats["unique_pools_seen"].add(pool_id)
                
                return token_mint, pool_id
                
            return None
                
        except Exception as e:
            logger.error(f"Error extracting token info: {e}")
            return None

    async def process_instruction(self, decoded: Dict[str, Any], signature: str, logs: List[str]):
        """Process a decoded instruction with thorough validation"""
        try:
            # Extract the amount and instruction type
            amount_sol = decoded.get("amount_sol", 0)
            instr_type = decoded.get("instruction_type", "unknown")
            
            # Special handling for V4 instructions (0x09)
            if instr_type == RAYDIUM_SWAP_INSTRUCTION_V4:
                logger.info(f"Processing V4 (0x09) instruction: {amount_sol:.6f} SOL")
                # V4 instructions might need different parameters
            
            # Skip amounts that are too small, which are often dust or errors
            if amount_sol < 0.01:
                logger.debug(f"Skipping tiny amount: {amount_sol:.8f} SOL")
                stats["filtered_by"]["size_too_small"] += 1
                return
            
            # Skip Jupiter transactions early
            if any("Jupiter" in log for log in logs):
                stats["filtered_by"]["jupiter"] += 1
                return
                
            # Extract token info
            token_info = self.extract_token_info(logs)
            if not token_info:
                stats["filtered_by"]["other"] += 1
                return
                
            token_mint, pool_id = token_info
            
            # IMPORTANT FIX: Allow whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc token
            # Skip other program IDs
            if token_mint in KNOWN_PROGRAM_IDS and token_mint != "whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc":
                logger.debug(f"Skipping program ID as token: {token_mint[:8]}...")
                stats["filtered_by"]["program_id"] += 1
                return
                
            # Skip tokens with LBUZKhRx prefix - this is the Meteora program
            if token_mint.startswith("LBUZKhRx"):
                logger.debug(f"Skipping Meteora program as token: {token_mint[:12]}...")
                stats["filtered_by"]["program_id"] += 1
                return
            
            # Skip stablecoins (simplified list for better success)
            if token_mint in STABLECOIN_ADDRESSES:
                logger.debug(f"Skipping stablecoin token: {token_mint[:8]}...")
                stats["filtered_by"]["stablecoin"] += 1
                return
            
            # Skip duplicated pool IDs as tokens
            if pool_id.startswith(token_mint[:8]) or token_mint.startswith(pool_id[:8]):
                logger.debug(f"Skipping likely duplicate address: {token_mint[:8]}...")
                stats["filtered_by"]["program_id"] += 1
                return
            
            # Check if this is a trade we're interested in
            if await self.is_actionable_trade(decoded, signature, logs, token_mint):
                # Verify pool security
                security_result = await self.verify_pool_security(pool_id, token_mint)
                if not security_result["is_valid"]:
                    logger.warning(f"Pool security verification failed for {token_mint[:8]}...")
                    stats["filtered_by"]["other"] += 1
                    return
                
                # Evaluate trade with GPU
                evaluation = await self.gpu.evaluate_trade(
                    token_mint=token_mint,
                    amount_sol=amount_sol,
                    instruction_type=instr_type
                )
                
                estimated_profit = evaluation["estimated_profit"]
                
                # Skip if below profit threshold with a buffer
                if estimated_profit < self.min_profit * 1.2:
                    logger.debug(f"Skipping low profit opportunity: {estimated_profit:.6f} SOL")
                    stats["filtered_by"]["profit_threshold"] += 1
                    return
                
                # Create opportunity
                opportunity = {
                    "decoded": decoded,
                    "signature": signature,
                    "token_mint": token_mint,
                    "pool_id": pool_id,
                    "amount_sol": amount_sol,
                    "timestamp": time.time(),
                    "instruction_type": instr_type,
                    "logs": logs,
                    "security_result": security_result,
                    "estimated_profit": estimated_profit
                }
                
                # Add to opportunity queue
                stats["opportunities"] += 1
                opportunity_queue.put(opportunity)
                
                logger.info(f"Found opportunity: {token_mint[:8]}... {amount_sol:.4f} SOL (est. profit: {estimated_profit:.6f} SOL)")
                
        except Exception as e:
            logger.error(f"Error processing instruction: {e}")
            logger.error(traceback.format_exc())
    
    async def is_actionable_trade(self, decoded: Dict[str, Any], signature: str, logs: List[str], token_mint: str = None) -> bool:
        """Determine if a trade is actionable with comprehensive validation"""
        # Check if we're already in a position
        if self.current_position is not None:
            return False
        
        # Check if trade is within size limits
        amount_sol = decoded.get("amount_sol", 0)
        if amount_sol < self.min_victim_size:
            stats["filtered_by"]["size_too_small"] += 1
            return False
        if amount_sol > self.max_victim_size:
            stats["filtered_by"]["size_too_large"] += 1
            return False
        
        # Check trading interval to prevent rapid trading
        current_time = time.time()
        if current_time - self.last_execution_time < self.min_trade_interval:
            stats["filtered_by"]["other"] += 1
            return False
        
        # Check if we are at max active trades
        if len(active_trades) >= 1:  # Limit to 1 active trade
            stats["filtered_by"]["other"] += 1
            return False
            
        # Check WSOL balance is sufficient
        wsol_balance = balance_updates.get("WSOL", 0)
        if wsol_balance < self.front_run_amount + 0.01:  # Add buffer for fees
            logger.warning(f"Insufficient WSOL balance: {wsol_balance:.6f} < {self.front_run_amount + 0.01:.6f}")
            stats["filtered_by"]["other"] += 1
            return False
        
        # Skip Jupiter transactions
        if any("Jupiter" in log for log in logs):
            stats["filtered_by"]["jupiter"] += 1
            return False
            
        # Skip LBUZKhRx tokens - additional check
        if token_mint and token_mint.startswith("LBUZKhRx"):
            logger.debug(f"Skipping Meteora program as token in actionable check: {token_mint[:12]}...")
            stats["filtered_by"]["program_id"] += 1
            return False
        
        # Estimate profit
        try:
            instr_type = decoded.get("instruction_type", 0)
            evaluation = await self.gpu.evaluate_trade(
                token_mint=token_mint or signature[:32],  # Use signature as proxy if no token_mint
                amount_sol=amount_sol,
                instruction_type=instr_type
            )
            
            estimated_profit = evaluation["estimated_profit"]
            if estimated_profit < self.min_profit:
                stats["filtered_by"]["profit_threshold"] += 1
                return False
        except Exception as e:
            logger.error(f"Error estimating profit: {e}")
            stats["filtered_by"]["other"] += 1
            return False
        
        return True
    
    async def verify_pool_security(self, pool_id: str, token_mint: str) -> Dict[str, Any]:
        """Verify pool security (mint revocation, LP locks) - simplified to allow most trades"""
        try:
            # Special handling for the whirLbMi token that caused issues
            if token_mint == "whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc":
                security_result = {
                    "is_valid": True,  # Always valid for this specific token
                    "mint_revoked": True,
                    "lp_locked": True,
                    "exit_liquidity": 15.0,
                    "has_sufficient_exit_liquidity": True,
                    "liquidity_lock_percent": 99.0,
                    "timestamp": time.time(),
                    "type": "whirlpool_special"
                }
                return security_result
                
            # Check cache first
            now = time.time()
            if pool_id in self.pool_security_cache:
                cache_entry = self.pool_security_cache[pool_id]
                cache_age = now - cache_entry["timestamp"]
                
                # Use cache if still valid (configurable seconds)
                cache_seconds = int(self.config.get("SECURITY_CACHE_SECONDS", 600))
                if cache_age < cache_seconds:
                    logger.debug(f"Using cached pool verification for {pool_id[:8]}...")
                    return cache_entry
                    
            # Try to use Rust binding for pool verification if available
            try:
                pool_info = await self.rate_limited_call(
                    self.trading_core.verify_pool_security_py(pool_id, token_mint)
                )
                
                if hasattr(pool_info, 'is_valid'):
                    # Return the security info using the Rust binding result
                    security_result = {
                        "is_valid": pool_info.is_valid,
                        "mint_revoked": pool_info.mint_authority_revoked,
                        "lp_locked": pool_info.liquidity_lock_percent >= self.min_liquidity_lock,
                        "exit_liquidity": pool_info.base_reserve,
                        "has_sufficient_exit_liquidity": pool_info.base_reserve >= self.min_exit_liquidity,
                        "liquidity_lock_percent": pool_info.liquidity_lock_percent,
                        "timestamp": now,
                        "type": "rust_binding"
                    }
                    
                    # Cache the result
                    self.pool_security_cache[pool_id] = security_result
                    
                    return security_result
            except Exception as e:
                logger.debug(f"Error using Rust binding for pool verification: {e}")
                # Fall through to default values below
            
            # Simplified default security values - assume most pools are valid
            security_result = {
                "is_valid": True,  # Assuming valid by default to increase success rate
                "mint_revoked": True,
                "lp_locked": True,
                "exit_liquidity": 15.0,  # Default exit liquidity
                "has_sufficient_exit_liquidity": True,
                "liquidity_lock_percent": 99.0,
                "timestamp": now,
                "type": "default"
            }
            
            # Cache the result
            self.pool_security_cache[pool_id] = security_result
            
            return security_result
            
        except Exception as e:
            logger.error(f"Error verifying pool security: {e}")
            return {
                "is_valid": True,  # Default to true to increase success rate
                "mint_revoked": True,
                "lp_locked": True,
                "exit_liquidity": 15.0,
                "has_sufficient_exit_liquidity": True,
                "liquidity_lock_percent": 99.0,
                "timestamp": time.time(),
                "error": str(e),
                "type": "error"
            }
            
    async def verify_token_account_exists(self, token_mint: str) -> bool:
        """Verify that the token account exists or can be created"""
        try:
            # Check if token account exists
            token_ata = await self.rate_limited_call(
                self.trading_core.get_token_ata_py(token_mint)
            )
            return bool(token_ata)
        except Exception:
            return False
    
    async def check_transaction_status(self, signature: str, token_mint: str = None):
        """Check status of a transaction with proper error handling"""
        try:
            # Wait for transaction confirmation
            status = await self.rate_limited_call(
                self.trading_core.get_transaction_status(signature)
            )
            
            if status and hasattr(status, 'confirmed'):
                if status.confirmed:
                    logger.info(f"Transaction confirmed: {signature[:8]}...")
                else:
                    logger.warning(f"Transaction not confirmed: {signature[:8]}...")
        except Exception as e:
            logger.error(f"Error checking transaction status: {e}")
    
    async def execute_trade(self, opportunity: Dict[str, Any]):
        """Execute a front-running trade with comprehensive validation"""
        async with self.trade_lock:
            # Extract key information
            token_mint = opportunity["token_mint"]
            amount_sol = opportunity["amount_sol"]
            signature = opportunity["signature"]
            pool_id = opportunity["pool_id"]
            security_result = opportunity["security_result"]
            
            # Double-check we're not in a position
            if self.current_position is not None:
                logger.info(f"Skipping trade - already in position")
                return
            
            logger.info(f"Executing trade for {token_mint[:8]}... ({amount_sol:.4f} SOL)")
            
            try:
                # Add to active trades
                active_trades.add(signature)
                
                # Update stats
                stats["executions"] += 1
                self.last_execution_time = time.time()
                
                # Make sure the token ATA exists before the swap - try multiple times
                token_ata = None
                ata_creation_success = False
                
                for attempt in range(3):
                    try:
                        token_ata = await self.rate_limited_call(
                            self.trading_core.ensure_token_ata_exists_py(token_mint)
                        )
                        if token_ata:
                            logger.info(f"Ensured token ATA exists: {token_ata[:8]}...")
                            ata_creation_success = True
                            break
                    except Exception as e:
                        logger.warning(f"Error ensuring token ATA exists (attempt {attempt+1}): {e}")
                        await asyncio.sleep(0.5)
                
                if not ata_creation_success:
                    logger.warning("Failed to create token ATA after multiple attempts, continuing anyway")
                
                # Get initial token balance (for verification later)
                initial_token_balance = 0.0
                try:
                    initial_token_balance = await self.rate_limited_call(
                        self.trading_core.get_token_balance_py(token_mint)
                    )
                    logger.info(f"Initial token balance: {initial_token_balance:.6f}")
                except Exception as e:
                    logger.debug(f"No initial token balance (expected): {e}")
                
                # Create swap parameters for entry (buy)
                try:
                    # Ensure adequate WSOL balance before swap
                    wsol_balance = balance_updates.get("WSOL", 0)
                    if wsol_balance < self.front_run_amount * 1.1:
                        logger.warning(f"Low WSOL balance: {wsol_balance:.6f}, may need wrapping")
                    
                    # Use higher priority fee for entry to ensure it gets picked up
                    priority_fee = int(self.config.get("PRIORITY_FEE", 950000)) * 2
                    
                    swap_params = self.trading_core.create_swap_params(
                        token=token_mint,
                        amount=self.front_run_amount,
                        is_buy=True,  # Buying token
                        slippage_bps=self.slippage_bps,
                        compute_limit=int(self.config.get("COMPUTE_LIMIT", 50000)),
                        priority_fee=priority_fee
                    )
                except Exception as e:
                    logger.error(f"Failed to create swap params: {e}")
                    logger.error(traceback.format_exc())
                    return
                
                # Execute the swap using trading core
                status = None
                try:
                    logger.info(f"Executing swap for {self.front_run_amount} SOL...")
                    status = await self.rate_limited_call(
                        self.trading_core.execute_swap(swap_params)
                    )
                except Exception as e:
                    logger.error(f"Failed to execute trade: {e}")
                    stats["trades_failed"] += 1
                    return
                
                # Check if swap was successful
                if hasattr(status, 'success') and status.success:
                    logger.info(f"Entry transaction successful: {status.signature[:8]}...")
                    
                    # Start verification of transaction
                    asyncio.create_task(self.check_transaction_status(status.signature, token_mint))
                    
                    # Wait longer for transaction confirmation - critical for tokens to arrive
                    logger.info(f"Waiting {ENTRY_MIN_WAIT_TIME}s for tokens to arrive...")
                    await asyncio.sleep(ENTRY_MIN_WAIT_TIME)
                    
                    # Token ATA handling with advanced verification
                    token_ata = None
                    ata_exists = False
                    
                    # Try multiple methods to get/create token ATA
                    for ata_attempt in range(3):
                        try:
                            token_ata = await self.rate_limited_call(
                                self.trading_core.get_token_ata_py(token_mint)
                            )
                            if token_ata:
                                logger.info(f"Retrieved token ATA: {token_ata[:8]}...")
                                ata_exists = True
                                break
                        except Exception as e:
                            logger.warning(f"Error getting token ATA (attempt {ata_attempt+1}): {e}")
                            await asyncio.sleep(0.5)
                    
                    # If we couldn't get the ATA, try to create it explicitly
                    if not ata_exists and token_ata is None:
                        try:
                            logger.info("Creating token ATA explicitly...")
                            token_ata = await self.rate_limited_call(
                                self.trading_core.ensure_token_ata_exists_py(token_mint)
                            )
                            logger.info(f"Created token ATA: {token_ata[:8]}...")
                            ata_exists = True
                        except Exception as e:
                            logger.error(f"Failed to create token ATA: {e}")
                
                    # Enhanced token balance verification
                    token_balance = 0.0
                    token_received = False
                    max_verify_attempts = 5  # Increased from original
                    
                    for attempt in range(max_verify_attempts):
                        try:
                            # Try to get token balance through Rust binding
                            token_balance = await self.rate_limited_call(
                                self.trading_core.get_token_balance_py(token_mint)
                            )
                            logger.info(f"Attempt {attempt+1}: Token balance: {token_balance:.6f} tokens")
                            
                            # ANY positive token balance is considered successful
                            if token_balance > 0:
                                token_received = True
                                logger.info(f"Token balance confirmed: {token_balance:.6f} tokens")
                                break
                            else:
                                # If we still don't see tokens, try again after waiting
                                logger.warning(f"No tokens detected yet. Waiting...")
                                await asyncio.sleep(1.0 + (attempt * 0.5))  # Increasing wait with each attempt
                        except Exception as e:
                            logger.warning(f"Failed to get token balance (attempt {attempt+1}): {e}")
                            await asyncio.sleep(1.0)
                    
                    # FIX: Even if we don't see tokens, we'll assume we got them and store a small token balance
                    # This handles cases where the balance reporting is delayed but tokens were received
                    if not token_received:
                        logger.warning(f"No tokens visibly received after {max_verify_attempts} attempts. Proceeding with token balance 0.0001")
                        token_balance = 0.0001  # Small amount to trigger exit
                        token_received = True
                    
                    # Store current position for exit
                    self.current_position = {
                        "token_mint": token_mint,
                        "pool_id": pool_id,
                        "entry_signature": status.signature,
                        "entry_time": time.time(),
                        "entry_amount": self.front_run_amount,
                        "security_result": security_result,
                        "victim_signature": signature,
                        "victim_amount": amount_sol,
                        "token_ata": token_ata,
                        "token_balance": token_balance,  # Store token balance
                        "exit_attempts": 0,
                        "exit_task": None  # Will store the scheduled exit task
                    }
                    
                    # Schedule exit after delay in case we don't see the victim tx
                    exit_task = asyncio.create_task(self.schedule_exit(5.0))  # 5 second timeout
                    if self.current_position is not None:
                        self.current_position["exit_task"] = exit_task
                        logger.info("Position established, exit scheduled")
                else:
                    error_msg = str(getattr(status, 'error', "Unknown error")) if status else "No status returned"
                    logger.error(f"Entry failed: {error_msg}")
                    stats["trades_failed"] += 1
            
            except Exception as e:
                logger.error(f"Error executing trade: {e}")
                logger.error(traceback.format_exc())
                stats["trades_failed"] += 1
            finally:
                # Remove from active trades
                if signature in active_trades:
                    active_trades.remove(signature)

    async def schedule_exit(self, delay_seconds: float):
        """Schedule an exit if not already exited"""
        try:
            await asyncio.sleep(delay_seconds)
            # Directly call exit_position instead of using command queue
            # This avoids thread/event loop issues
            if self.current_position is not None:
                position_copy = self.current_position.copy()
                await self.exit_position(position_copy)
        except asyncio.CancelledError:
            # Task was cancelled - this is normal when exiting early
            logger.debug("Exit task cancelled")
        except Exception as e:
            logger.error(f"Error in scheduled exit: {e}")
            
            # If we hit an error, try via command queue as fallback
            if self.current_position is not None:
                command_queue.put(("exit_position", self.current_position))

    async def exit_position(self, position: Dict[str, Any]):
        """Exit a position with robust error handling and retry logic"""
        if not position:
            return
        
        # Prevent concurrent exits
        async with self.trade_lock:
            # Check if we're still in this position
            if self.current_position is None or self.current_position.get("entry_signature") != position.get("entry_signature"):
                return
            
            token_mint = position["token_mint"]
            entry_time = position["entry_time"]
            
            # Track exit attempts
            position["exit_attempts"] = position.get("exit_attempts", 0) + 1
            
            # Calculate time in position
            time_in_position = time.time() - entry_time
            
            logger.info(f"Exiting position for {token_mint[:8]}... (time in position: {time_in_position:.2f}s)")
            
            try:
                # Avoid rapid selling if too soon (RPC rate limit protection)
                if time_in_position < 1.5:  # Increased from 0.8
                    wait_time = 1.5 - time_in_position
                    logger.info(f"Waiting {wait_time:.2f}s before exiting position")
                    await asyncio.sleep(wait_time)
                
                # Enhanced token balance retrieval for exit
                token_balance = 0.0
                balance_attempts = 3
                
                # Try multiple times to get the token balance
                for attempt in range(balance_attempts):
                    try:
                        token_balance = await self.rate_limited_call(
                            self.trading_core.get_token_balance_py(token_mint)
                        )
                        logger.info(f"Exit token balance (attempt {attempt+1}): {token_balance:.6f} tokens")
                        if token_balance > 0:
                            break
                        await asyncio.sleep(1.0)
                    except Exception as e:
                        logger.warning(f"Error getting exit token balance: {e}")
                        await asyncio.sleep(0.5)
                
                # If we still don't have a balance, try using the stored value or fallback
                if token_balance <= 0:
                    # NEW FIX: Use a minimum value to attempt an exit even if we don't see tokens
                    # This often happens with new tokens where the balance reporting is delayed
                    token_balance = 0.0001  # Small amount to trigger exit
                    logger.warning(f"No token balance detected, using minimal value for exit: {token_balance}")
                    
                # Record WSOL balance before exiting
                wsol_balance_before = balance_updates.get("WSOL", 0)

                # Use very aggressive slippage for this emergency exit
                slippage_bps = 5000  # 50% slippage

                # Higher priority fee for emergency exits
                priority_fee = 3000000  # Very high priority

                try:
                    # Create emergency exit swap params
                    emergency_swap_params = self.trading_core.create_swap_params(
                        token=token_mint,
                        amount=token_balance,
                        is_buy=False,  # Selling token
                        slippage_bps=slippage_bps,
                        compute_limit=int(self.config.get("COMPUTE_LIMIT", 50000)),
                        priority_fee=priority_fee
                    )

                    # Attempt emergency exit
                    logger.info(f"Attempting emergency exit with minimal token amount and {slippage_bps/100}% slippage")
                    status = await self.rate_limited_call(
                        self.trading_core.execute_swap(emergency_swap_params)
                    )

                    if hasattr(status, 'success') and status.success:
                        logger.info(f"Emergency exit successful: {status.signature[:8]}...")

                        # Clear position and success
                        if self.current_position and self.current_position.get("exit_task"):
                            try:
                                self.current_position["exit_task"].cancel()
                            except Exception:
                                pass

                        self.current_position = None
                        return
                except Exception as e:
                    logger.error(f"Emergency exit failed: {e}")

                # Record WSOL balance before exiting
                wsol_balance_before = balance_updates.get("WSOL", 0)

                # Increase slippage for each exit attempt - be even more aggressive
                base_slippage = 2000  # Start with 20% slippage
                slippage_bps = base_slippage + (position["exit_attempts"] * 800)  # Add 8% per retry

                # Log the exit attempt details
                logger.info(f"Attempting to sell {token_balance:.6f} tokens with {slippage_bps/100:.1f}% slippage (attempt {position['exit_attempts']})")

                # Higher priority fee for exits to ensure they go through
                priority_fee = 2000000 + (position["exit_attempts"] * 500000)  # Significantly higher priority

                # Create swap parameters for exit (sell)
                try:
                    swap_params = self.trading_core.create_swap_params(
                        token=token_mint,
                        amount=token_balance,  # Sell entire balance
                        is_buy=False,  # Selling token
                        slippage_bps=slippage_bps,  # Higher slippage for exits
                        compute_limit=int(self.config.get("COMPUTE_LIMIT", 50000)),
                        priority_fee=priority_fee  # Higher priority fee
                    )
                except Exception as e:
                    logger.error(f"Failed to create swap params: {e}")
                    logger.error(traceback.format_exc())
                    
                    # Try once more with different values if this fails
                    try:
                        swap_params = self.trading_core.create_swap_params(
                            token=token_mint,
                            amount=token_balance * 0.99,  # Try slightly less
                            is_buy=False,
                            slippage_bps=3000,  # Much higher slippage
                            compute_limit=int(self.config.get("COMPUTE_LIMIT", 50000)),
                            priority_fee=2500000  # Very high priority fee
                        )
                    except Exception as retry_e:
                        logger.error(f"Failed to create swap params on retry: {retry_e}")
                        
                        # Schedule another exit attempt
                        if position["exit_attempts"] < self.max_exit_retries:
                            exit_task = asyncio.create_task(self.schedule_exit(3.0))
                            if self.current_position:
                                self.current_position["exit_task"] = exit_task
                        else:
                            self.current_position = None
                        return
                
                # Execute the swap with rate limiting
                try:
                    status = await self.rate_limited_call(
                        self.trading_core.execute_swap(swap_params)
                    )
                except Exception as e:
                    logger.error(f"RPC error executing swap: {e}")
                    
                    # Schedule retry
                    if position["exit_attempts"] < self.max_exit_retries:
                        retry_delay = min(2.0 * position["exit_attempts"], 5.0)
                        exit_task = asyncio.create_task(self.schedule_exit(retry_delay))
                        if self.current_position:
                            self.current_position["exit_task"] = exit_task
                    else:
                        self.current_position = None
                    return

                if hasattr(status, 'success') and status.success:
                    logger.info(f"Exit successful: {status.signature[:8]}...")

                    # Wait for balance updates
                    await asyncio.sleep(POST_TRANSACTION_WAIT_TIME)

                    # Calculate profit based on WSOL balance change
                    wsol_balance_after = balance_updates.get("WSOL", 0)
                    wsol_received = wsol_balance_after - wsol_balance_before

                    # If websocket didn't update, let's try to calculate profit
                    if wsol_received == 0:
                        # Try to update balance directly
                        try:
                            wsol_balance_after = await self.rate_limited_call(
                                self.trading_core.get_token_balance_py(WSOL_ADDRESS)
                            )
                            wsol_received = wsol_balance_after - wsol_balance_before
                            logger.info(f"Updated WSOL balance directly: {wsol_balance_after:.6f} SOL")
                        except Exception as e:
                            logger.error(f"Failed to update balance directly: {e}")
                            wsol_received = 0.01  # Assume small positive result for stats

                    # Calculate net profit
                    net_profit = wsol_received - self.front_run_amount
                    
                    if net_profit > 0:
                        logger.info(f"Profit: {net_profit:.6f} SOL (tx: {status.signature[:8]}...)")
                        stats["total_profit"] += net_profit
                        stats["last_profit"] = time.time()
                        stats["trades_successful"] += 1
                    else:
                        logger.warning(f"Loss: {net_profit:.6f} SOL (tx: {status.signature[:8]}...)")
                        stats["trades_failed"] += 1

                    # Clear position and cancel any pending exit tasks
                    if self.current_position and self.current_position.get("exit_task"):
                        try:
                            self.current_position["exit_task"].cancel()
                        except Exception:
                            pass
                    
                    self.current_position = None
                    
                    # Check if we should update balances
                    await self.check_balances()
                else:
                    error_msg = str(getattr(status, 'error', "Unknown error")) if status else "No status returned"
                    logger.error(f"Exit failed: {error_msg}")
                    
                    # Schedule retry if we haven't exceeded max attempts
                    if position["exit_attempts"] < self.max_exit_retries:
                        retry_delay = min(2.0 * position["exit_attempts"], 5.0)
                        exit_task = asyncio.create_task(self.schedule_exit(retry_delay))
                        if self.current_position:
                            self.current_position["exit_task"] = exit_task
                    else:
                        # Give up after max retries
                        logger.warning(f"Giving up on exit after {position['exit_attempts']} attempts")
                        self.current_position = None
            
            except Exception as e:
                logger.error(f"Error exiting position: {e}")
                logger.error(traceback.format_exc())
                
                # Schedule retry if we haven't exceeded max attempts
                if position["exit_attempts"] < self.max_exit_retries:
                    retry_delay = min(2.0 * position["exit_attempts"], 5.0)
                    exit_task = asyncio.create_task(self.schedule_exit(retry_delay))
                    if self.current_position:
                        self.current_position["exit_task"] = exit_task
                else:
                    # Give up after max retries
                    logger.warning(f"Giving up on exit after {position['exit_attempts']} attempts")
                    self.current_position = None

    def execution_worker(self):
        """Worker thread for processing execution queue"""
        logger.info("Execution worker thread started")
        
        # Create a new event loop for this thread
        worker_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(worker_loop)
        
        while bot_running.is_set():
            try:
                # Get opportunity from queue with timeout
                try:
                    opportunity = opportunity_queue.get(timeout=0.5)
                except queue.Empty:
                    continue
                
                # Check for commands
                try:
                    cmd, data = command_queue.get_nowait()
                    if cmd == "shutdown":
                        logger.info("Execution worker received shutdown command")
                        return
                    elif cmd == "exit_position" and self.current_position is not None:
                        # Execute in this thread's event loop
                        worker_loop.run_until_complete(self.exit_position(data))
                    elif cmd == "check_exit" and self.current_position is not None:
                        # Victim transaction detected, exit our position
                        worker_loop.run_until_complete(self.exit_position(self.current_position))
                except queue.Empty:
                    pass
                
                # Process opportunity
                if self.current_position is None:
                    # Execute trade in this thread's event loop
                    worker_loop.run_until_complete(self.execute_trade(opportunity))
                
            except Exception as e:
                logger.error(f"Error in execution worker: {e}")
                logger.error(traceback.format_exc())
        
        # Clean up resources
        worker_loop.close()
    
    def stats_worker(self):
        """Worker thread for printing stats"""
        logger.info("Stats worker thread started")
        
        # Create a new event loop for this thread if needed for any async operations
        worker_loop = asyncio.new_event_loop()
        asyncio.set_event_loop(worker_loop)
        
        while bot_running.is_set():
            try:
                # Sleep for 60 seconds
                time.sleep(60)
                
                # Calculate runtime
                runtime = time.time() - stats["start_time"]
                hours = int(runtime // 3600)
                minutes = int((runtime % 3600) // 60)
                seconds = int(runtime % 60)
                
                # Print stats
                logger.info(f"=== Stats (Runtime: {hours:02d}:{minutes:02d}:{seconds:02d}) ===")
                logger.info(f"Logs processed: {stats['logs_processed']}")
                logger.info(f"Raydium logs found: {stats['ray_logs_found']}")
                logger.info(f"Instructions parsed: {stats['instructions_parsed']}")
                logger.info(f"Instruction types: V3 (0x03): {stats['instruction_types']['0x03']}, " +
                            f"V4 (0x09): {stats['instruction_types']['0x09']}, " +
                            f"V6 (0x06): {stats['instruction_types']['0x06']}")
                logger.info(f"Opportunities: {stats['opportunities']}")
                logger.info(f"Executions: {stats['executions']}")
                logger.info(f"Successful trades: {stats['trades_successful']}")
                logger.info(f"Failed trades: {stats['trades_failed']}")
                logger.info(f"Total profit: {stats['total_profit']:.6f} SOL")
                
                # Print balance info
                sol_balance = balance_updates.get("SOL", 0)
                wsol_balance = balance_updates.get("WSOL", 0)
                logger.info(f"SOL balance: {sol_balance:.6f} SOL")
                logger.info(f"WSOL balance: {wsol_balance:.6f} SOL")
                
                # Print unique pools
                logger.info(f"Unique pools seen: {len(stats['unique_pools_seen'])}")
                
            except Exception as e:
                logger.error(f"Error in stats worker: {e}")
                
        # Clean up resources
        worker_loop.close()


def main():
    """Main entry point with platform-aware signal handling"""
    try:
        # Get event loop
        loop = asyncio.get_event_loop()
        
        # Setup signal handlers (platform-aware)
        if platform.system() != "Windows":
            # Unix-like systems support this signal handling
            signals = (signal.SIGTERM, signal.SIGINT)
            for s in signals:
                loop.add_signal_handler(
                    s, lambda s=s: asyncio.create_task(shutdown(s, loop))
                )
        
        # Set thread pool size
        loop.set_default_executor(
            concurrent.futures.ThreadPoolExecutor(max_workers=16)
        )
        
        # Create and run bot
        bot = RaydiumMEVBot()
        loop.run_until_complete(bot.run())
        
    except KeyboardInterrupt:
        logger.info("Keyboard interrupt received, shutting down")
    except Exception as e:
        logger.critical(f"Fatal error: {e}")
        logger.critical(traceback.format_exc())
    finally:
        logger.info("Bot shutdown complete")


async def shutdown(signal, loop):
    """Shutdown coroutine with proper cleanup"""
    logger.info(f"Received exit signal {signal.name}")
    tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    
    # Cancel all tasks
    for task in tasks:
        task.cancel()
    
    # Wait for all tasks to complete
    await asyncio.gather(*tasks, return_exceptions=True)
    
    # Stop the event loop
    loop.stop()


if __name__ == "__main__":
    main() fix the script why doesnt it work properly for raydium solana pairs only ?

Okay, so the user is asking why the provided script doesn't work properly for Raydium Solana pairs. Let me start by understanding the script's purpose and where it might be going wrong.

First, the script is a MEV bot targeting Raydium swaps on Solana. It's supposed to detect swap transactions in the mempool and front-run them. The user says it's not working for SOL pairs, so maybe it's missing SOL-specific handling.

Looking at the code, I notice that when extracting token info, the script might not account for SOL being involved. Raydium pairs can be between SOL and another token, but SOL isn't an SPL token. The script uses WSOL (wrapped SOL) for transactions. So, if the victim transaction is swapping SOL for a token, the bot needs to recognize WSOL in the logs.

In the `extract_token_info` function, the code looks for token mints and pool IDs. But if the swap is SOL to a token, the token mint would be the other token, and the base asset is SOL. The script might be missing cases where one side of the pair is SOL/WSOL. For example, when the victim is swapping SOL for a token, the logs might mention WSOL's address (So1111...), which is a known program ID. The current code skips known program IDs, which would include WSOL, leading to the bot ignoring those transactions.

Another point is in the `verify_pool_security` function. If the pool is a SOL pair, the checks might need to handle WSOL differently. The script's current security checks might be treating WSOL as a regular token, but it's a special case.

Also, in the `execute_trade` function, when creating swap parameters, the bot uses the token mint. For SOL pairs, the token mint should be WSOL's address when swapping SOL. The script might not be correctly identifying when the trade involves SOL and thus not using WSOL appropriately.

The `is_actionable_trade` function checks the WSOL balance, which is correct for SOL pairs. But if the victim's transaction is SOL to token, the bot's logic to detect the token mint might be failing because it's excluding WSOL as a known program ID. So the token extraction step is skipping the correct token mint (the other token) because the pool might be misidentified.

To fix this, the `extract_token_info` function needs to correctly identify when one of the tokens in the pair is WSOL. It should allow WSOL as a valid token mint in the swap. Additionally, when the victim's swap involves SOL, the bot should recognize that and handle WSOL correctly in the swap parameters.

Another area to check is the `KNOWN_PROGRAM_IDS` set. WSOL's address (So1111...) is included here, which might cause the script to skip swaps involving SOL. The fix would be to remove WSOL from the known program IDs that are excluded, allowing it to be treated as a valid token mint when encountered in swap logs.

Also, in the `verify_pool_security` function, there's a special case for whirlpool, but maybe SOL pools need a similar exception. The script should recognize SOL pools and handle their security checks appropriately, possibly skipping some checks that don't apply to WSOL.

In the `execute_trade` function, when creating swap parameters for SOL pairs, the token should be WSOL, and the bot should ensure that the swap is correctly structured as a SOL-to-token or token-to-SOL trade.

So, the main issues are:

1. WSOL is being excluded as a known program ID, causing the bot to ignore SOL pairs.

2. The token extraction logic isn't correctly identifying SOL/WSOL as one side of the pair.

3. The swap parameters might not be correctly using WSOL when necessary.

To address these, the fixes would involve:

- Removing WSOL from the `KNOWN_PROGRAM_IDS` so it's not excluded.

- Modifying the `extract_token_info` function to recognize when WSOL is part of the swap pair.

- Adjusting the swap parameter creation to handle WSOL correctly, ensuring that swaps involving SOL are properly executed.

(Due to technical issues, the search service is temporarily unavailable.)

To fix the script to properly handle Raydium SOL pairs, we need to make key adjustments to handle SOL/WSOL transactions correctly. Here are the essential changes:
diff
Copy

# Step 1: Remove WSOL from KNOWN_PROGRAM_IDS to allow SOL pair trading
KNOWN_PROGRAM_IDS = {
    RAYDIUM_PROGRAM_ID,
    TOKEN_PROGRAM_ID,
    SYSTEM_PROGRAM_ID,
    ATA_PROGRAM_ID,
    COMPUTE_BUDGET_PROGRAM_ID,
    SERUM_PROGRAM_ID,
    "LBUZKhRx4amdUADHbwgRxh3h4F6FX2rXKj5uCQVGTFT",  # Meteora DLMM Program
-    "So11111111111111111111111111111111111111112",  # WSOL Address
    "CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK",  # Raydium Concentrated Liquidity
    "9W959DqEETiGZocYWCQPaJ6sBmUzgfxXfqGeTEdp3aQP",  # Orca Whirlpool
    "JUP4Fb2cqiRUcaTHdrPC8h2gNsA2ETXiPDD33WcGuJB",   # Jupiter Aggregator
    "Jupiter1111111111111111111111111111111",         # Jupiter alias
    "BONK9xNjA3527soNjSuYSRY1YLRdJsD8a7zJxdYMXTFF"   # BONK token
}

# Step 2: Update extract_token_info to handle SOL pairs
def extract_token_info(self, logs: List[str]) -> Optional[Tuple[str, str]]:
    try:
        # Add explicit WSOL handling
        wsol_address = WSOL_ADDRESS
        
        # Look for WSOL in logs
        wsol_in_logs = any(wsol_address in log for log in logs)
        
        # Existing code...
        for log in logs:
            # Add special case for WSOL mentions
            if wsol_address in log:
                wsol_in_logs = True
                
        # Modified token detection logic
        if wsol_in_logs:
            # Look for the non-WSOL token in the pair
            for word in potential_tokens:
                if word != wsol_address and word not in KNOWN_PROGRAM_IDS:
                    token_mint = word
                    break
                    
            # Try to find pool ID containing "amm" or "pool"
            for addr in potential_addresses:
                if ("amm" in log.lower() or "pool" in log.lower()) and addr != wsol_address:
                    pool_id = addr
                    break

        # Rest of existing logic...
        
        # Add explicit WSOL pair handling
        if token_mint == wsol_address:
            # Look for alternate token in logs
            for word in potential_tokens:
                if word != wsol_address and word not in KNOWN_PROGRAM_IDS:
                    token_mint = word
                    break

        return token_mint, pool_id

# Step 3: Modify swap parameter creation for SOL pairs
async def execute_trade(self, opportunity: Dict[str, Any]):
    # Add WSOL pair detection
    is_sol_pair = False
    if opportunity["token_mint"] == WSOL_ADDRESS:
        is_sol_pair = True
        # Find the actual token mint from logs
        token_info = self.extract_token_info(opportunity["logs"])
        if token_info:
            opportunity["token_mint"] = token_info[0]

    # When creating swap parameters
    if is_sol_pair:
        # For SOL pairs, we need to handle WSOL wrapping
        try:
            # Ensure WSOL account exists
            await self.trading_core.ensure_token_ata_exists_py(WSOL_ADDRESS)
            
            # Create special swap params for SOL pair
            swap_params = self.trading_core.create_swap_params(
                token=opportunity["token_mint"],
                amount=self.front_run_amount,
                is_buy=True,
                sol_pair=True,  # Add this parameter to your Rust code
                slippage_bps=self.slippage_bps,
                compute_limit=int(self.config.get("COMPUTE_LIMIT", 50000)),
                priority_fee=priority_fee
            )
        except Exception as e:
            logger.error(f"SOL pair swap error: {e}")

# Step 4: Update Rust bindings to handle SOL pairs
"""
In your Rust code (solana_rust_bot/src/lib.rs), add sol_pair parameter handling:

#[pyclass]
struct SwapParams {
    #[pyo3(get, set)]
    sol_pair: bool,
    // ... other fields
}

#[pymethods]
impl RaydiumTradingCore {
    fn create_swap_params(
        &self,
        token: String,
        amount: f64,
        is_buy: bool,
        sol_pair: bool,
        slippage_bps: u64,
        compute_limit: u32,
        priority_fee: u64
    ) -> PyResult<SwapParams> {
        // Handle SOL pairs differently in swap construction
        if sol_pair {
            // Add WSOL account handling and construct SOL pair swap
        }
    }
}
"""

These changes address the main issues with SOL pair handling:

    Allows WSOL to be recognized as a valid trading pair

    Improves token detection logic for SOL-based swaps

    Adds special handling for SOL pair transactions in both Python and Rust code

    Ensures proper WSOL account management

The script should now properly detect and execute trades for SOL pairs on Raydium. Make sure to update your Rust bindings with the SOL pair handling logic and recompile with maturin develop --release.

Note: This requires corresponding changes to the Rust implementation to properly handle SOL pair swaps, including WSOL account management and transaction construction specific to SOL pairs.
New chat
AI-generated, for reference only