please add # In advanced_trading.py
class AdvancedTrading:
    def __init__(self):
        self._metrics = {
            'trades_executed': 0,
            'total_volume': 0.0,
            'successful_trades': 0,
            'failed_trades': 0,
            'average_latency': 0.0,
            'peak_latency': 0.0,
            'profit_loss': 0.0
        }
        
        # Add process queue
        self.process_queue = asyncio.Queue(maxsize=int(os.getenv('QUEUE_SIZE', 1000)))

    def _save_metrics(self):
        """Save trading metrics"""
        try:
            metrics_data = {
                **self._metrics,
                'timestamp': time.time(),
                'session_duration': time.time() - self._start_time
            }
            logger.info(f"Trading Metrics: {json.dumps(metrics_data, indent=2)}")
        except Exception as e:
            logger.error(f"Error saving metrics: {e}")

    async def cleanup(self):
        try:
            self._save_metrics()
            
            # Clear process queue
            while not self.process_queue.empty():
                try:
                    self.process_queue.get_nowait()
                except asyncio.QueueEmpty:
                    break
                    
            logger.info("Advanced trading cleanup completed")
        except Exception as e:
            logger.error(f"Advanced trading cleanup error: {e}")   to ----> this    >>>> import asyncio
import base64
import json
import logging
import os
import time
from typing import Dict, List, Optional
from decimal import Decimal
import torch
import torch.cuda.amp as amp
from dataclasses import dataclass, field
import psutil
import websockets
from concurrent.futures import ThreadPoolExecutor
from dotenv import load_dotenv
from config_loader import config

# Load environment variables from .env file
load_dotenv()

logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

@dataclass
class TradingMetrics:
    messages_received: int = 0
    opportunities_found: int = 0
    successful_trades: int = 0
    total_profit: Decimal = Decimal('0')
    start_time: float = field(default_factory=time.time)
    sandwich_profits: Decimal = Decimal('0')
    impact_profits: Decimal = Decimal('0')
    failed_attempts: Dict[str, int] = field(default_factory=dict)
    
    strategy_performance: Dict[str, Dict] = field(default_factory=lambda: {
        'sandwich': {'success': 0, 'fail': 0, 'avg_profit': Decimal('0'), 'avg_time': 0.0},
        'impact': {'success': 0, 'fail': 0, 'avg_profit': Decimal('0'), 'avg_time': 0.0}
    })
    
    execution_stats: Dict[str, List[float]] = field(default_factory=lambda: {
        'latency': [], 'gas_used': [], 'slippage': []
    })

class AdvancedTrading:
    def __init__(self, trader, config):
        self.trader = trader
        self.config = config
        self.active = False
        self.metrics = TradingMetrics()

        # GPU setup with tensor cores enabled
        self.device = torch.device('cuda')
        torch.backends.cuda.matmul.allow_tf32 = True
        torch.backends.cudnn.benchmark = True
        
        # Initialize CUDA streams with priorities
        self.streams = {
            'sandwich': torch.cuda.Stream(priority=-2),  # Highest priority
            'impact': torch.cuda.Stream(priority=-1),    # High priority
            'compute': torch.cuda.Stream(priority=0)     # Normal priority
        }
        
        # Initialize thread pool for CPU tasks
        cpu_count = psutil.cpu_count(logical=False)
        self.thread_pool = ThreadPoolExecutor(max_workers=cpu_count)
        
        # Setup processing buffers
        self.setup_processing_buffers()
        
        # Initialize opportunity queue with prioritization
        self.opportunity_queue = asyncio.PriorityQueue()
        
        # Load trading parameters from environment variables
        self._load_trading_parameters()
        
        logger.info("Advanced trading initialized with GPU acceleration")

    def _load_trading_parameters(self):
        """Load and validate trading parameters"""
        try:
            # Load from environment variables
            self.MIN_PROFIT_PCT = float(os.getenv("MIN_PROFIT_PCT"))
            self.MAX_PROFIT_PCT = float(os.getenv("MAX_PROFIT_PCT"))
            self.MAX_PRICE_IMPACT = float(os.getenv("MAX_PRICE_IMPACT"))
            self.MAX_POSITION_SIZE = float(os.getenv("POSITION_MAX_SIZE"))
            self.INITIAL_SIZE = float(os.getenv("POSITION_INITIAL_SIZE"))
            self.GAS_RESERVE = float(os.getenv("GAS_RESERVE"))
            self.SLIPPAGE = float(os.getenv("MAX_SPREAD"))
            
            # Sandwich specific parameters
            self.MIN_SANDWICH_SIZE = float(os.getenv("SANDWICH_MIN_VICTIM_SIZE"))  # Minimum victim tx size in SOL
            self.MAX_SANDWICH_IMPACT = float(os.getenv("SANDWICH_MAX_IMPACT"))  # Maximum impact for sandwich trades
            self.SANDWICH_PROFIT_THRESHOLD = float(os.getenv("SANDWICH_MIN_PROFIT"))  # Minimum profit threshold
            
            logger.info("Trading parameters loaded successfully")
            
        except KeyError as e:
            logger.error(f"Missing configuration parameter: {e}")
            raise

    def setup_processing_buffers(self):
        """Initialize optimized GPU processing buffers"""
        try:
            gpu_mem = torch.cuda.get_device_properties(0).total_memory
            buffer_size = min(2048, gpu_mem // (1024 * 1024 * 10))
            
            # Create CPU tensors first
            self.opportunity_buffer = torch.zeros(
                (buffer_size, 128), 
                dtype=torch.float16,
                device='cpu'  # Create on CPU first
            ).pin_memory()  # Pin the memory
            
            self.impact_buffer = torch.zeros(
                (buffer_size // 2, 2), 
                dtype=torch.float32,
                device='cpu'
            ).pin_memory()
            
            self.pattern_buffer = torch.zeros(
                (1024, 64),
                dtype=torch.float16,
                device='cpu'
            ).pin_memory()

            # Move to GPU if available
            if torch.cuda.is_available():
                with torch.cuda.stream(self.streams['compute']):
                    self.opportunity_buffer = self.opportunity_buffer.to(
                        device=self.device,
                        non_blocking=True
                    )
                    self.impact_buffer = self.impact_buffer.to(
                        device=self.device,
                        non_blocking=True
                    )
                    self.pattern_buffer = self.pattern_buffer.to(
                        device=self.device,
                        non_blocking=True
                    )
                    torch.cuda.synchronize()
                    logger.debug("GPU buffers initialized")
                    
        except Exception as e:
            logger.error(f"Failed to initialize GPU buffers: {e}")
            raise
    async def _check_sandwich_opportunity(self, raw_data: bytes) -> Optional[Dict]:
        """Check for sandwich opportunities with enhanced MEV protection"""
        try:
            with torch.cuda.stream(self.streams['sandwich']):
                # Decode and validate victim transaction
                victim_tx = await self._decode_victim_transaction(raw_data)
                if not victim_tx or victim_tx['amount'] < self.MIN_SANDWICH_SIZE:
                    return None

                # Calculate optimal positions
                positions = await self._calculate_optimal_positions(
                    victim_tx['pool'],
                    victim_tx['amount'],
                    victim_tx['direction']
                )
                if not positions:
                    return None

                # MEV protection check
                is_safe = await self.trader.mev_protection.verify_sandwich(
                    positions['front_run'],
                    victim_tx['amount'],
                    positions['back_run']
                )
                if not is_safe:
                    return None

                # Calculate expected profit
                profit = await self._calculate_sandwich_profit(positions, victim_tx['pool'])
                if profit < self.SANDWICH_PROFIT_THRESHOLD:
                    return None

                return {
                    'type': 'sandwich',
                    'profit': profit,
                    'positions': positions,
                    'victim': victim_tx,
                    'pool': victim_tx['pool'],
                    'confidence': 0.9,
                    'priority': 1
                }

        except Exception as e:
            logger.error(f"Sandwich check error: {e}")
            return None

    async def _calculate_optimal_positions(self, pool_address: str, victim_amount: float, direction: bool) -> Optional[Dict]:
        """Calculate optimal front-run and back-run positions"""
        try:
            with torch.cuda.stream(self.streams['sandwich']):
                # Get pool state
                pool_state = await self.trader.get_pool_state(pool_address)
                if not pool_state:
                    return None
                    
                # Convert to tensors for GPU calculation
                base_reserve = torch.tensor(float(pool_state.base_reserve), device=self.device)
                quote_reserve = torch.tensor(float(pool_state.quote_reserve), device=self.device)
                victim_amount_tensor = torch.tensor(victim_amount, device=self.device)
                
                # Calculate optimal front-run size
                with amp.autocast(device_type='cuda'):
                    # Start with 50% of victim tx size
                    front_run_size = victim_amount_tensor * 0.5
                    
                    # Iteratively optimize front-run size
                    for _ in range(5):  # Max 5 iterations
                        # Calculate price impact
                        if direction:  # Victim buying
                            new_quote_reserve = quote_reserve + front_run_size
                            new_base_reserve = base_reserve * quote_reserve / new_quote_reserve
                            price_impact = (new_base_reserve / base_reserve - 1).abs()
                        else:  # Victim selling
                            new_base_reserve = base_reserve + front_run_size
                            new_quote_reserve = quote_reserve * base_reserve / new_base_reserve
                            price_impact = (new_quote_reserve / quote_reserve - 1).abs()
                        
                        # Adjust size based on impact
                        if price_impact > self.MAX_PRICE_IMPACT:
                            front_run_size *= 0.9
                        else:
                            break
                    
                    # Calculate back-run size
                    back_run_size = front_run_size * 1.1  # Slightly larger to account for slippage
                    
                    # Calculate minimum output amounts with slippage buffer
                    front_run_min_out = self._calculate_min_output(
                        float(front_run_size),
                        float(base_reserve),
                        float(quote_reserve),
                        self.SLIPPAGE
                    )
                    
                    back_run_min_out = self._calculate_min_output(
                        float(back_run_size),
                        float(new_base_reserve),
                        float(new_quote_reserve),
                        self.SLIPPAGE
                    )
                    
                    return {
                        'front_run': float(front_run_size),
                        'back_run': float(back_run_size),
                        'front_run_min_out': front_run_min_out,
                        'back_run_min_out': back_run_min_out,
                        'price_impact': float(price_impact)
                    }

        except Exception as e:
            logger.error(f"Position calculation error: {e}")
            return None

    async def _execute_sandwich_trade(self, opportunity: Dict, priority_fee: int) -> bool:
        """Execute sandwich trade with precise timing"""
        try:
            # Execute front-run with higher priority
            front_run_fee = int(priority_fee * 1.2)  # 20% higher for front-run
            front_run_result = await self.trader.trading_core.execute_trade(
                pool_address=opportunity['pool'],
                amount_in=opportunity['positions']['front_run'],
                min_amount_out=opportunity['positions']['front_run_min_out'],
                priority_fee=front_run_fee,
                compute_limit=600000  # Higher compute limit for complex trades
            )

            if not front_run_result:
                logger.error("Front-run execution failed")
                return False

            # Minimal waiting period for victim transaction
            await asyncio.sleep(0.05)  # 50ms delay

            # Back-run with lower priority to save on fees
            back_run_fee = int(priority_fee * 0.8)  # 20% lower for back-run
            back_run_result = await self.trader.trading_core.execute_trade(
                pool_address=opportunity['pool'],
                amount_in=opportunity['positions']['back_run'],
                min_amount_out=opportunity['positions']['back_run_min_out'],
                priority_fee=back_run_fee
            )

            if not back_run_result:
                logger.error("Back-run execution failed")
                return False

            return True

        except Exception as e:
            logger.error(f"Sandwich execution error: {e}")
            return False

    def _calculate_min_output(self, amount_in: float, base_reserve: float, quote_reserve: float, slippage: float) -> float:
        """Calculate minimum output amount with slippage protection"""
        try:
            constant_product = base_reserve * quote_reserve
            new_quote_reserve = quote_reserve + amount_in
            ideal_output = base_reserve - (constant_product / new_quote_reserve)
            min_output = ideal_output * (1 - slippage)
            return max(min_output, 0)
        except Exception as e:
            logger.error(f"Min output calculation error: {e}")
            return 0

    async def _optimize_priority_fee(self, opportunity_type: str, profit: float) -> int:
        """Dynamic priority fee optimization"""
        try:
            base_fee = await self.trader.mev_protection.get_priority_fee()
            
            multipliers = {
                'sandwich': 1.5,   # Higher priority for sandwich
                'impact': 1.2      # Medium priority for impact trades
            }
            
            fee_multiplier = multipliers.get(opportunity_type, 1.0)
            if profit > float(os.getenv("PROFIT_THRESHOLD")):
                fee_multiplier *= float(os.getenv("FEE_MULTIPLIER"))

            priority_fee = int(base_fee * fee_multiplier)
            
            return min(
                max(priority_fee, int(os.getenv("MIN_PRIORITY_FEE"))),
                int(os.getenv("MAX_PRIORITY_FEE"))
            )

        except Exception as e:
            logger.error(f"Fee optimization error: {e}")
            return int(os.getenv("DEFAULT_PRIORITY_FEE"))

    async def execute_opportunity(self, opportunity: Dict) -> bool:
        """Execute opportunity with dynamic fee optimization"""
        start_time = time.time()
        
        try:
            if not await self._validate_execution_conditions(opportunity):
                return False

            priority_fee = await self._optimize_priority_fee(
                opportunity['type'],
                opportunity['profit']
            )

            if opportunity['type'] == 'sandwich':
                result = await self._execute_sandwich_trade(opportunity, priority_fee)
            else:  # impact trade
                result = await self._execute_impact_trade(opportunity, priority_fee)

            if result:
                execution_time = time.time() - start_time
                await self._update_execution_stats(opportunity, execution_time)
                
                self.metrics.successful_trades += 1
                if opportunity['type'] == 'sandwich':
                    self.metrics.sandwich_profits += Decimal(str(opportunity['profit']))
                else:
                    self.metrics.impact_profits += Decimal(str(opportunity['profit']))

                logger.info(f"Trade successful: {opportunity['type']}, Profit: {opportunity['profit']:.4f}")
                return True

            return False

        except Exception as e:
            logger.error(f"Execution error: {e}")
            execution_time = time.time() - start_time
            await self._update_failed_execution_stats(opportunity, execution_time)
            return False

    async def start_websocket_monitoring(self):
        """High-performance WebSocket monitoring"""
        self.active = True
        logger.info("Starting advanced trading monitor...")
        
        while self.active:
            try:
                async with websockets.connect(
                    os.getenv("HELIUS_WS_URL"),
                    ping_interval=int(os.getenv("WS_PING_INTERVAL")),
                    ping_timeout=int(os.getenv("WS_PING_TIMEOUT")),
                    close_timeout=int(os.getenv("WS_CLOSE_TIMEOUT")),
                    max_size=int(os.getenv("WS_MAX_MESSAGE_SIZE"))
                ) as ws:
                    await ws.send(json.dumps({
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "logsSubscribe",
                        "params": [
                            {"mentions": [self.config['programs']['raydium']]},
                            {"commitment": "processed"}
                        ]
                    }))
                    
                    processors = [
                        self._process_messages(ws),
                        self._monitor_network_conditions(),
                        self._optimize_execution_parameters()
                    ]
                    await asyncio.gather(*processors)
                    
            except websockets.exceptions.ConnectionClosed:
                logger.warning("WebSocket connection closed. Reconnecting...")
                await asyncio.sleep(0.1)
            except Exception as e:
                logger.error(f"WebSocket error: {e}")
                await asyncio.sleep(0.1)

    async def _process_messages(self, ws):
        """Process messages with GPU acceleration"""
        while self.active:
            try:
                msg = await ws.recv()
                self.metrics.messages_received += 1
                
                with amp.autocast(device_type='cuda'):
                    opportunities = await self._parallel_opportunity_check(msg)
                    if opportunities:
                        await self._execute_best_opportunity(opportunities)
                
            except Exception as e:
                logger.error(f"Message processing error: {e}")

    @torch.amp.autocast('cuda')
    async def _parallel_opportunity_check(self, msg: str) -> List[Dict]:
        """Check for sandwich and price impact opportunities in parallel"""
        try:
            data = json.loads(msg)
            if 'params' not in data:
                return []

            logs = data['params'].get('result', {}).get('value', {}).get('logs', [])
            opportunities = []

            for log in logs:
                if "ray_log:" not in log:
                    continue

                raw_data = base64.b64decode(log.split('ray_log: ')[1].strip())

                # Parallel opportunity checking
                with torch.cuda.stream(self.streams['compute']):
                    results = await asyncio.gather(
                        self._check_sandwich_opportunity(raw_data),
                        self._check_price_impact_opportunity(raw_data)
                    )

                    valid_results = [r for r in results if r and self._validate_opportunity(r)]
                    if valid_results:
                        opportunities.extend(self._calculate_opportunity_scores(valid_results))

            return opportunities

        except Exception as e:
            logger.error(f"Parallel check error: {e}")
            return []

    async def _check_price_impact_opportunity(self, raw_data: bytes) -> Optional[Dict]:
        """Check for price impact opportunities"""
        try:
            with torch.cuda.stream(self.streams['impact']):
                # Extract transaction data
                tx_data = await self._process_trade_data(raw_data)
                if not tx_data:
                    return None

                # Calculate impact and profit
                impact = await self._calculate_price_impact(
                    tx_data['amount_in'],
                    tx_data['amount_out'],
                    tx_data['pool']
                )
                if impact > self.MAX_PRICE_IMPACT:
                    return None

                profit = await self._calculate_impact_profit(tx_data, impact)
                if profit < self.MIN_PROFIT_PCT:
                    return None

                return {
                    'type': 'impact',
                    'profit': profit,
                    'amount': tx_data['amount_in'],
                    'pool': tx_data['pool'],
                    'impact': impact,
                    'confidence': 0.85,
                    'priority': 2
                }

        except Exception as e:
            logger.error(f"Impact check error: {e}")
            return None

    async def cleanup(self):
        """Cleanup resources properly"""
        logger.info("Starting cleanup...")
        try:
            self.active = False
            
            # Clean GPU resources
            torch.cuda.empty_cache()
            for stream in self.streams.values():
                stream.synchronize()
            
            # Save performance metrics
            self._save_metrics()
            
            # Close thread pool
            self.thread_pool.shutdown(wait=True)
            
            logger.info("Cleanup completed")
            
        except Exception as e:
            logger.error(f"Cleanup error: {e}")
            raise