Okay, let me try to figure out why the websocket isn't connecting in the first script but works in the second.

First, I'll compare the two scripts to see the differences in how they handle the websocket connections.

In the first script, the `monitor_mempool` method is supposed to connect to the websocket. The URL is built using `HELIUS_WS_URL` or `WS_URL`, and there's an API key added if present. The connection is within a `websockets.connect` context, and they subscribe to logs for the Raydium program ID. Then they process messages in a loop.

In the second script, the `monitor_ray_logs` method does something similar but also includes parameters like `ping_interval`, `ping_timeout`, and `close_timeout` when connecting. Also, the subscription message might be structured differently.

Looking closer, maybe the first script is missing some headers or parameters. For example, the second script includes `extra_headers` when connecting if there's an API key. The first script appends the API key as a query parameter, but maybe the service requires it in the headers instead. That's a common difference between services—some expect API keys in headers, others in the URL.

Another possibility is the handling of the subscription message. The first script's `monitor_mempool` sends a `logsSubscribe` with `{"mentions": [RAYDIUM_PROGRAM_ID]}`, while the second script's `monitor_ray_logs` does the same. But maybe the first script isn't waiting for the subscription confirmation properly. The second script explicitly waits for a response after sending the subscription message, which the first might not be doing, leading to missed messages.

Also, the first script's `process_transaction` checks for "Program log: Instruction: Swap" in the logs. But if the log format has changed or isn't exactly matching that string, the script might not detect the swap instructions. The second script uses a different method, decoding `ray_log` entries, which might be more reliable.

Another angle: the first script uses `asyncio.create_task` to run the websocket in the background but might not be properly handling reconnections or errors. The second script has a loop that retries connecting with increasing delays, which is more robust.

Additionally, the first script might not be sending pings or handling timeouts, leading to the connection dropping without reconnecting. The second script includes `ping_interval` and `ping_timeout`, which help maintain the connection.

Possible fixes for the first script:

1. Include API key in headers instead of the URL query.

2. Add ping parameters to the websocket connection.

3. Properly handle subscription confirmations.

4. Improve error handling and reconnection logic.

5. Ensure the logs are parsed correctly for the swap instruction.

For example, in the first script's `monitor_mempool`, when connecting, they should set headers if using an API key. Also, after subscribing, wait for the subscription confirmation message before proceeding.

Another thing: the first script's `process_transaction` looks for logs containing "Program log: Instruction: Swap". If the actual log message is different (like "Program log: Instruction: SwapV2" or similar), it would miss those. The second script decodes the instruction type from the log data, which is more precise.

So, the main issues could be:

- Incorrect API key handling (query vs headers)

- Missing websocket parameters (ping settings)

- Not waiting for subscription confirmation

- Incorrect log parsing for swap instructions

- Lack of proper reconnection logic

To fix the first script, I'd adjust the websocket connection to include headers for the API key, add ping parameters, handle subscription confirmations, and verify the log parsing logic is correctly identifying swap instructions.

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

To resolve the websocket connection issue in your first script, here are the key fixes needed:

    API Key in Headers Instead of Query Params

python
Copy

# In monitor_mempool() function:
headers = {}
if API_KEY:
    headers["api-key"] = API_KEY  # Helius expects API key in headers

async with websockets.connect(
    ws_url, 
    extra_headers=headers,  # Add headers here
    ping_interval=20,       # Add ping parameters
    ping_timeout=30,
    close_timeout=60
) as ws:

    Handle Subscription Confirmation

python
Copy

# After sending subscription request:
await ws.send(json.dumps({
    "jsonrpc": "2.0",
    "id": 1,
    "method": "logsSubscribe",
    "params": [
        {"mentions": [RAYDIUM_PROGRAM_ID]},
        {"commitment": "processed"}
    ]
}))

# Wait for subscription confirmation
response = await ws.recv()
confirmation = json.loads(response)
if "error" in confirmation:
    logger.error(f"Subscription failed: {confirmation['error']}")
    return

    Improved Log Parsing

python
Copy

# In process_transaction():
# Look for the actual instruction byte instead of log messages
for log in logs:
    if f"Program {RAYDIUM_PROGRAM_ID} invoke" in log:
        is_raydium_v4 = True
    
    if "Program data: " in log and is_raydium_v4:
        try:
            data = base64.b64decode(log.split("Program data: ")[1])
            if len(data) > 0 and data[0] == RAYDIUM_SWAP_INSTRUCTION:
                # Found swap instruction
                has_swap = True

    Add Reconnection Logic

python
Copy

# Wrap the websocket connection in a retry loop:
retry_count = 0
while self.is_running:
    try:
        async with websockets.connect(...) as ws:
            # Successful connection
            retry_count = 0
            # ... rest of logic ...
    except Exception as e:
        retry_count += 1
        delay = min(2 ** retry_count, 30)
        logger.error(f"Connection failed: {e}. Retrying in {delay}s")
        await asyncio.sleep(delay)

    Update Instruction Detection

python
Copy

# Use proper instruction parsing instead of log messages
def decode_raydium_instruction(data: bytes) -> Optional[Dict]:
    """Decode Raydium v4 swap instruction"""
    if len(data) < 1:
        return None
    
    instr_type = data[0]
    if instr_type != RAYDIUM_SWAP_INSTRUCTION:
        return None
    
    return {
        "instruction_type": instr_type,
        "amount_in": int.from_bytes(data[1:9], "little"),
        # ... parse other fields ...
    }

Full Fixed Connection Code
python
Copy

async def monitor_mempool(self):
    """Monitor mempool with proper connection handling"""
    retry_count = 0
    headers = {"api-key": API_KEY} if API_KEY else {}
    
    while self.is_running:
        try:
            async with websockets.connect(
                HELIUS_WS_URL or WS_URL,
                extra_headers=headers,
                ping_interval=20,
                ping_timeout=30,
                close_timeout=60
            ) as ws:
                retry_count = 0
                logger.info("Connected to mempool websocket")

                # Subscribe to Raydium logs
                sub_msg = {
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "logsSubscribe",
                    "params": [
                        {"mentions": [RAYDIUM_PROGRAM_ID]},
                        {"commitment": "processed"}
                    ]
                }
                await ws.send(json.dumps(sub_msg))
                
                # Verify subscription
                response = await ws.recv()
                sub_data = json.loads(response)
                if "error" in sub_data:
                    logger.error(f"Subscription error: {sub_data['error']}")
                    continue

                # Process messages
                async for msg in ws:
                    await self.process_transaction(msg)
                    
        except Exception as e:
            retry_count += 1
            delay = min(2 ** retry_count, 30)
            logger.error(f"Connection error: {e}. Retry in {delay}s")
            await asyncio.sleep(delay)

These changes will:

    Properly handle API keys in headers

    Add essential websocket keep-alive parameters

    Implement proper subscription confirmation

    Add exponential backoff for reconnections

    Use proper instruction decoding instead of log message scanning

The connection issues were likely caused by:

    Missing API key headers (Helius requires them in headers)

    No keep-alive pings leading to dropped connections

    Not handling initial subscription confirmation properly

    Fragile log message pattern matching instead of proper instruction decoding

Remember to also update your instruction parsing to properly handle the Raydium v4 swap instruction format instead of relying on log message strings.#!/usr/bin/env python3
"""
Raydium MEV Bot - Optimized for Raydium V4 AMM (0x09 instructions)
With GPU acceleration and Windows compatibility
"""
import asyncio
import base64
import json
import logging
import os
import sys
import time
import traceback
import websockets
import torch
import numpy as np
from typing import Dict, List, Any, Optional, Set, Tuple

# Import Rust bindings
from solana_rust_bot import (
    RaydiumTradingCore,
    SwapParams,
    TokenPairSetting,
    TradingConfig,
    TokenSafetyCheck,
    TransactionStatus
)

# Constants
LAMPORTS_PER_SOL = 1_000_000_000
RAYDIUM_SWAP_INSTRUCTION = 0x09  # V4 AMM instruction
RAYDIUM_PROGRAM_ID = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"
WSOL_ADDRESS = "So11111111111111111111111111111111111111112"

# Environment setup
RPC_URL = os.getenv("RPC_URL", "https://winny-rychu7-fast-mainnet.helius-rpc.com")
WS_URL = os.getenv("WS_URL", "wss://winny-rychu7-fast-mainnet.helius-rpc.com")
HELIUS_WS_URL = os.getenv("HELIUS_WS_URL", "wss://winny-rychu7-fast-mainnet.helius-rpc.com")
API_KEY = os.getenv("API_KEY", "86ab5634-df30-4a5b-bcfb-3f53d7095ca2")
KEYPAIR_PATH = r"C:\solana_rust_bot\keypair.bin"  # Windows path with r-string
FRONT_RUN_AMOUNT = float(os.getenv("FRONT_RUN_AMOUNT", "0.3"))
MIN_PROFIT = float(os.getenv("MIN_PROFIT", "0.002"))
SLIPPAGE_BPS = int(os.getenv("SLIPPAGE_BPS", "1000"))

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(message)s',
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("raydium_mev")

class GPUAccelerator:
    """GPU acceleration for high-performance calculations"""
    
    def __init__(self):
        """Initialize GPU with optimized settings for 4070 Ti"""
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        
        # Configure CUDA settings for maximum performance
        if torch.cuda.is_available():
            # Enable TF32 for faster computation
            torch.backends.cuda.matmul.allow_tf32 = True
            torch.backends.cudnn.benchmark = True
            torch.backends.cudnn.deterministic = False
            
            # Create CUDA streams for concurrent operations
            self.stream = torch.cuda.Stream(priority=-3)
            
            # Configure for FP16 precision (faster)
            self.dtype = torch.float16
            
            logger.info(f"GPU acceleration enabled: {torch.cuda.get_device_name(0)}")
            logger.info(f"CUDA version: {torch.version.cuda}")
            logger.info(f"Available memory: {torch.cuda.get_device_properties(0).total_memory / 1024 / 1024 / 1024:.2f} GB")
        else:
            self.dtype = torch.float32
            logger.warning("GPU not available, using CPU")
            
        # Initialize tensors for trading calculations
        self.initialize_tensors()
    
    def initialize_tensors(self):
        """Initialize tensors with optimized size (176)"""
        # Prices and amounts tensors (176 is optimized size)
        self.prices = torch.zeros(176, dtype=self.dtype, device=self.device)
        self.amounts = torch.zeros(176, dtype=self.dtype, device=self.device)
        
        # Warm up the GPU
        if torch.cuda.is_available():
            with torch.cuda.stream(self.stream):
                for _ in range(1000):
                    _ = torch.matmul(self.prices, self.amounts)
                torch.cuda.synchronize()
    
    @torch.compile(fullgraph=True)
    def calculate_price_impact(self, pool_reserves: Tuple[float, float], amount_in: float, 
                             is_buy: bool) -> float:
        """Calculate price impact using GPU acceleration"""
        # Convert to tensors
        reserves_tensor = torch.tensor(pool_reserves, dtype=self.dtype, device=self.device)
        amount_tensor = torch.tensor([amount_in], dtype=self.dtype, device=self.device)
        
        # Calculate constant product formula (optimized)
        k = reserves_tensor[0] * reserves_tensor[1]
        
        if is_buy:
            new_reserve_in = reserves_tensor[0] + amount_tensor
            new_reserve_out = k / new_reserve_in
            amount_out = reserves_tensor[1] - new_reserve_out
            price_impact = 1.0 - ((amount_out / amount_tensor) * (reserves_tensor[0] / reserves_tensor[1]))
        else:
            new_reserve_in = reserves_tensor[1] + amount_tensor
            new_reserve_out = k / new_reserve_in
            amount_out = reserves_tensor[0] - new_reserve_out
            price_impact = 1.0 - ((amount_out / amount_tensor) * (reserves_tensor[1] / reserves_tensor[0]))
        
        return price_impact.item()
    
    def estimate_profit(self, pool_reserves: Tuple[float, float], amount_in: float, 
                       victim_amount: float) -> float:
        """Estimate front-running profit using GPU calculations"""
        # Calculate price before victim's trade
        if torch.cuda.is_available():
            with torch.cuda.stream(self.stream):
                # Step 1: Our front-run buy
                impact1 = self.calculate_price_impact(pool_reserves, amount_in, True)
                buy_amount_out = amount_in * (1 - impact1)
                
                # New pool state after our buy
                new_pool_token = pool_reserves[0] + amount_in
                new_pool_sol = pool_reserves[1] - buy_amount_out
                
                # Step 2: Victim's trade - they buy after us, at worse price
                impact2 = self.calculate_price_impact((new_pool_token, new_pool_sol), victim_amount, True)
                
                # Step 3: Our sell after victim's trade
                new_pool_token += victim_amount
                new_pool_sol -= victim_amount * (1 - impact2)
                
                # Sell our tokens
                impact3 = self.calculate_price_impact((new_pool_token, new_pool_sol), buy_amount_out, False)
                sell_amount_out = buy_amount_out * (1 - impact3)
                
                # Calculate profit
                profit = sell_amount_out - amount_in
                
                return profit
        else:
            # Simple CPU calculation as fallback
            return victim_amount * 0.005  # Estimated 0.5% profit

class BalanceMonitor:
    """Monitor balance changes via websocket subscription"""
    
    def __init__(self, wallet_pubkey: str, token_accounts: Dict[str, str]):
        """Initialize balance monitor"""
        self.wallet_pubkey = wallet_pubkey
        self.token_accounts = token_accounts  # Map of token mint to token account
        self.sol_balance = 0.0
        self.token_balances = {}
        self.callbacks = []
        self.is_running = False
        self.task = None
        
        # Track historical balances for profit calculation
        self.balance_history = []
    
    def register_callback(self, callback):
        """Register a callback for balance changes"""
        self.callbacks.append(callback)
    
    async def start(self, initial_sol: float, initial_tokens: Dict[str, float]):
        """Start monitoring with initial balances"""
        self.sol_balance = initial_sol
        self.token_balances = initial_tokens.copy()
        self.is_running = True
        
        # Add initial balance snapshot
        self.balance_history.append({
            'timestamp': time.time(),
            'sol': self.sol_balance,
            'tokens': self.token_balances.copy()
        })
        
        # Start websocket subscription in background
        self.task = asyncio.create_task(self.subscribe_to_balance_changes())
        
        return self.sol_balance, self.token_balances
    
    async def stop(self):
        """Stop monitoring"""
        self.is_running = False
        if self.task:
            self.task.cancel()
            try:
                await self.task
            except asyncio.CancelledError:
                pass
    
    async def subscribe_to_balance_changes(self):
        """Subscribe to balance changes via websocket"""
        ws_url = HELIUS_WS_URL if HELIUS_WS_URL else WS_URL
        
        # Add API key if provided
        if API_KEY and "?" not in ws_url:
            ws_url += f"?api-key={API_KEY}"
        
        reconnect_attempts = 0
        
        while self.is_running:
            try:
                async with websockets.connect(ws_url) as ws:
                    logger.info("Connected to balance monitoring websocket")
                    reconnect_attempts = 0
                    
                    # Subscribe to SOL balance changes
                    await ws.send(json.dumps({
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "accountSubscribe",
                        "params": [
                            self.wallet_pubkey,
                            {"encoding": "jsonParsed", "commitment": "processed"}
                        ]
                    }))
                    
                    # Subscribe to token account balance changes
                    account_subscriptions = []
                    for token_mint, token_account in self.token_accounts.items():
                        await ws.send(json.dumps({
                            "jsonrpc": "2.0",
                            "id": len(account_subscriptions) + 2,
                            "method": "accountSubscribe",
                            "params": [
                                token_account,
                                {"encoding": "jsonParsed", "commitment": "processed"}
                            ]
                        }))
                        account_subscriptions.append(token_account)
                    
                    # Process incoming messages
                    while self.is_running:
                        msg = await ws.recv()
                        await self.process_balance_update(msg)
                        
            except (websockets.exceptions.ConnectionClosed, 
                    websockets.exceptions.WebSocketException) as e:
                if not self.is_running:
                    break
                    
                reconnect_attempts += 1
                backoff = min(5 * reconnect_attempts, 30)
                logger.error(f"Balance websocket error: {e}. "
                            f"Reconnecting in {backoff} seconds...")
                await asyncio.sleep(backoff)
                
            except Exception as e:
                if not self.is_running:
                    break
                    
                logger.error(f"Unexpected error in balance monitor: {e}")
                logger.debug(traceback.format_exc())
                await asyncio.sleep(5)
    
    async def process_balance_update(self, msg: str):
        """Process balance update from websocket"""
        try:
            data = json.loads(msg)
            
            # Skip if not a valid notification
            if not ("params" in data and "result" in data["params"]):
                return
                
            result = data["params"]["result"]
            value = result.get("value", {})
            
            # Process SOL balance update
            if "lamports" in value:
                lamports = value["lamports"]
                new_sol_balance = lamports / LAMPORTS_PER_SOL
                
                if new_sol_balance != self.sol_balance:
                    old_balance = self.sol_balance
                    self.sol_balance = new_sol_balance
                    logger.info(f"SOL balance changed: {old_balance:.4f} -> {new_sol_balance:.4f}")
                    
                    # Add to history
                    self.balance_history.append({
                        'timestamp': time.time(),
                        'sol': self.sol_balance,
                        'tokens': self.token_balances.copy()
                    })
                    
                    # Notify callbacks
                    for callback in self.callbacks:
                        await callback('sol', old_balance, new_sol_balance)
            
            # Process token balance update
            elif "data" in value and "parsed" in value["data"]:
                parsed = value["data"]["parsed"]
                if "info" in parsed and "tokenAmount" in parsed["info"]:
                    token_amount = parsed["info"]["tokenAmount"]
                    mint = parsed["info"].get("mint")
                    
                    if mint and "uiAmount" in token_amount:
                        amount = token_amount["uiAmount"]
                        
                        if mint in self.token_balances and amount != self.token_balances[mint]:
                            old_amount = self.token_balances[mint]
                            self.token_balances[mint] = amount
                            logger.info(f"Token {mint} balance changed: {old_amount:.4f} -> {amount:.4f}")
                            
                            # Add to history
                            self.balance_history.append({
                                'timestamp': time.time(),
                                'sol': self.sol_balance,
                                'tokens': self.token_balances.copy()
                            })
                            
                            # Notify callbacks
                            for callback in self.callbacks:
                                await callback('token', old_amount, amount, mint)
                                
        except json.JSONDecodeError:
            pass  # Not valid JSON
        except Exception as e:
            logger.error(f"Error processing balance update: {e}")
    
    def calculate_profit(self, window_seconds: float = 60.0) -> float:
        """Calculate profit over a time window"""
        if not self.balance_history:
            return 0.0
            
        now = time.time()
        # Find the balance snapshot closest to window_seconds ago
        start_idx = 0
        for i, snapshot in enumerate(self.balance_history):
            if now - snapshot['timestamp'] <= window_seconds:
                start_idx = max(0, i - 1)
                break
        
        if start_idx >= len(self.balance_history) - 1:
            return 0.0
            
        start = self.balance_history[start_idx]
        end = self.balance_history[-1]
        
        # Calculate SOL value change
        sol_change = end['sol'] - start['sol']
        
        # Calculate token value change (simplified)
        # In a real implementation, you'd convert token amounts to SOL value
        
        return sol_change

class RaydiumMEVBot:
    """MEV bot for front-running Raydium swaps with GPU acceleration"""
    
    def __init__(self):
        """Initialize MEV bot with Rust trading core and GPU acceleration"""
        logger.info("Initializing Raydium MEV Bot with GPU acceleration...")
        
        # Initialize trading core with correct keypair path
        self.trading_core = RaydiumTradingCore(
            rpc_url=RPC_URL,
            keypair_path=r"C:\solana_rust_bot\keypair.bin"
        )
        
        # Get wallet public key
        self.wallet_pubkey = self.trading_core.get_public_key()
        logger.info(f"Wallet public key: {self.wallet_pubkey}")
        
        # Initialize GPU accelerator
        self.gpu = GPUAccelerator()
        
        # Track active trades and state
        self.active_trades: Set[str] = set()
        self.seen_txs: Set[str] = set()
        self.current_position = None
        self.token_accounts: Dict[str, str] = {}
        
        # Statistics tracking
        self.stats = {
            'success_count': 0,
            'failure_count': 0,
            'profit_total': 0.0,
            'last_profit': 0.0,
            'trades_today': 0,
            'start_time': time.time()
        }
        
        # Balance monitor
        self.balance_monitor = None
        
        # Initialize with healthcheck
        self.is_running = True
        self.shutdown_event = asyncio.Event()
        
        # Create cache for token safety checks
        self.token_safety_cache = {}
        
    async def initialize(self):
        """Initialize the bot with initial balance check and setup"""
        # Check balances
        balances = await self.trading_core.get_balances()
        sol_balance, token_balances = balances
        
        logger.info(f"Initial SOL Balance: {sol_balance:.4f}")
        
        # Get token accounts for balance monitoring
        self.token_accounts = {}
        for token_mint, amount in token_balances.items():
            # Skip zero balances
            if amount <= 0:
                continue
                
            # Ensure ATA exists and get the address
            ata = await self.trading_core.ensure_ata(token_mint)
            self.token_accounts[token_mint] = ata
            logger.info(f"Token {token_mint}: {amount:.4f} (Account: {ata})")
        
        # Ensure WSOL ATA exists
        wsol_ata = await self.trading_core.ensure_ata(WSOL_ADDRESS)
        self.token_accounts[WSOL_ADDRESS] = wsol_ata
        logger.info(f"WSOL Account: {wsol_ata}")
        
        # Initialize balance monitor
        self.balance_monitor = BalanceMonitor(self.wallet_pubkey, self.token_accounts)
        
        # Register callback for balance changes
        self.balance_monitor.register_callback(self.handle_balance_change)
        
        # Start balance monitor with initial balances
        await self.balance_monitor.start(sol_balance, token_balances)
        
        # Start healthcheck in background
        asyncio.create_task(self.healthcheck())
        
        return True
        
    async def handle_balance_change(self, balance_type: str, old_value: float, 
                                  new_value: float, token_mint: str = None):
        """Handle balance change notification"""
        # If we have an open position and a token balance dropped to near zero, it indicates a successful exit
        if self.current_position and balance_type == 'token' and token_mint == self.current_position['token_mint']:
            if new_value < 0.01 and old_value > 0.01:
                logger.info(f"Detected position exit from balance change. Token {token_mint} sold.")
                
                # Calculate profit
                profit = self.balance_monitor.calculate_profit(window_seconds=60.0)
                logger.info(f"Estimated profit: {profit:.6f} SOL")
                
                # Update stats
                self.stats['last_profit'] = profit
                self.stats['profit_total'] += profit
                
                # Clear current position
                self.current_position = None
        
    async def healthcheck(self):
        """Periodic healthcheck to verify system state"""
        while self.is_running:
            try:
                uptime = time.time() - self.stats['start_time']
                hours = uptime / 3600
                
                logger.info(f"Bot status: Running for {hours:.2f} hours | "
                           f"Active trades: {len(self.active_trades)} | "
                           f"Position: {'Open' if self.current_position else 'None'} | "
                           f"Success/Fail: {self.stats['success_count']}/{self.stats['failure_count']} | "
                           f"Profit: {self.stats['profit_total']:.6f} SOL")
                
                # Check GPU health if using CUDA
                if torch.cuda.is_available():
                    mem_used = torch.cuda.memory_allocated() / 1024 / 1024 / 1024
                    logger.info(f"GPU memory usage: {mem_used:.2f} GB")
                
                await asyncio.sleep(30)
            except Exception as e:
                logger.error(f"Healthcheck error: {e}")
                await asyncio.sleep(10)

    async def monitor_mempool(self):
        """Monitor mempool for Raydium swap transactions using Helius websocket"""
        logger.info(f"Connecting to mempool monitoring at {HELIUS_WS_URL or WS_URL}")
        
        # Use the enhanced Helius websocket for better performance
        ws_url = HELIUS_WS_URL if HELIUS_WS_URL else WS_URL
        
        # Add API key if provided
        if API_KEY and "?" not in ws_url:
            ws_url += f"?api-key={API_KEY}"
        
        reconnect_attempts = 0
        
        while self.is_running:
            try:
                async with websockets.connect(ws_url) as ws:
                    logger.info("Connected to mempool websocket")
                    reconnect_attempts = 0
                    
                    # Subscribe to Raydium program logs - enhanced subscription
                    await ws.send(json.dumps({
                        "jsonrpc": "2.0",
                        "id": 1,
                        "method": "logsSubscribe",
                        "params": [
                            {"mentions": [RAYDIUM_PROGRAM_ID]},
                            {"commitment": "processed"}
                        ]
                    }))
                    
                    response = await ws.recv()
                    logger.info(f"Subscription established")
                    
                    # Process incoming messages
                    while self.is_running:
                        msg = await ws.recv()
                        await self.process_transaction(msg)
            
            except (websockets.exceptions.ConnectionClosed, 
                    websockets.exceptions.WebSocketException) as e:
                if not self.is_running:
                    break
                    
                reconnect_attempts += 1
                backoff = min(5 * reconnect_attempts, 30)
                logger.error(f"Websocket connection error: {e}. "
                            f"Reconnecting in {backoff} seconds...")
                await asyncio.sleep(backoff)
                
            except Exception as e:
                if not self.is_running:
                    break
                    
                logger.error(f"Unexpected error in websocket connection: {e}")
                logger.error(traceback.format_exc())
                await asyncio.sleep(5)

    async def process_transaction(self, msg: str):
        """Process incoming transaction data with GPU acceleration"""
        try:
            data = json.loads(msg)
            
            # Check if this is a proper logs notification
            if not ("params" in data and "result" in data["params"]):
                return
                
            logs = data.get('params', {}).get('result', {}).get('logs', [])
            signature = data.get('params', {}).get('result', {}).get('signature', '')
            
            if not logs or signature in self.seen_txs:
                return
            
            self.seen_txs.add(signature)
            
            # Check if this is a transaction we sent
            if signature in self.active_trades:
                logger.info(f"Received confirmation for our transaction: {signature}")
                return
                
            # Skip if we already have a position
            if self.current_position:
                return
                
            # Look for Raydium swap instruction
            for log in logs:
                if "Program log: Instruction: Swap" in log and f"Program {RAYDIUM_PROGRAM_ID}" in logs:
                    # Extract swap details
                    swap_data = await self.parse_swap_instruction(logs, signature)
                    
                    if swap_data and await self.is_profitable(swap_data):
                        await self.execute_front_run(swap_data, signature)
                    break

        except json.JSONDecodeError:
            pass  # Not valid JSON
        except Exception as e:
            logger.error(f"Processing error: {e}")
            logger.debug(traceback.format_exc())

    async def parse_swap_instruction(self, logs: List[str], signature: str) -> Optional[Dict]:
        """Extract swap details from transaction logs"""
        try:
            # Extract token mint from the logs
            token_mint = None
            amount_in_sol = None
            accounts = []
            
            # Flags to track status
            is_raydium_v4 = False
            has_instruction_9 = False
            
            for log in logs:
                # Look for Raydium V4 AMM signatures
                if "Program 675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8 invoke" in log:
                    is_raydium_v4 = True
                
                # Check instruction type from data    
                if "Program data: " in log and is_raydium_v4:
                    try:
                        data = base64.b64decode(log.split("Program data: ")[1])
                        if len(data) > 0 and data[0] == 9:  # Instruction 0x09
                            has_instruction_9 = True
                            
                            # Extract amount if data is long enough
                            if len(data) >= 9:
                                amount_lamports = int.from_bytes(data[1:9], byteorder='little')
                                amount_in_sol = amount_lamports / LAMPORTS_PER_SOL
                    except:
                        pass
                
                # Extract token mint looking at token program logs
                if "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" in log and "create" not in log.lower():
                    parts = log.split()
                    for i, part in enumerate(parts):
                        if part == "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA":
                            # The token mint is typically 2 parts before a program mention
                            if i >= 2:
                                potential_mint = parts[i-2]
                                # Verify it's a valid pubkey format
                                if len(potential_mint) in (43, 44) and " " not in potential_mint:
                                    if token_mint is None:  # First occurrence is often the token mint
                                        token_mint = potential_mint
                
                # Extract all account pubkeys
                if len(log.strip()) in (43, 44) and " " not in log.strip():
                    accounts.append(log.strip())
            
            # Skip if not a valid Raydium V4 swap
            if not (is_raydium_v4 and has_instruction_9):
                return None
                
            # Skip if we couldn't identify the token
            if not token_mint:
                return None
                
            # Normalize SOL address to wSOL address if needed
            if token_mint == "11111111111111111111111111111111":
                token_mint = WSOL_ADDRESS
                
            # Use default amount if we couldn't extract it
            if not amount_in_sol:
                amount_in_sol = 0.1  # Default
            
            return {
                "token_mint": token_mint,
                "amount_in_sol": amount_in_sol,
                "accounts": accounts,
                "signature": signature
            }
            
        except Exception as e:
            logger.error(f"Error parsing swap instruction: {e}")
            return None

    async def is_profitable(self, swap_data: Dict) -> bool:
        """Determine if a front-running opportunity is profitable using GPU acceleration"""
        try:
            token_mint = swap_data.get("token_mint")
            victim_amount = swap_data.get("amount_in_sol", 0.1)
            
            # Skip transactions with missing data
            if not token_mint:
                return False
                
            # Check token safety
            # Use cached result if available (for 5 minutes)
            current_time = time.time()
            if token_mint in self.token_safety_cache:
                cache_time, safety = self.token_safety_cache[token_mint]
                if current_time - cache_time < 300:  # 5 minutes
                    if not safety.is_safe:
                        return False
                else:
                    # Cache expired, remove it
                    del self.token_safety_cache[token_mint]
            
            # Not in cache or expired, perform safety check
            if token_mint not in self.token_safety_cache:
                safety = await self.trading_core.check_token_safety(token_mint)
                self.token_safety_cache[token_mint] = (current_time, safety)
                
                if not safety.is_safe:
                    logger.warning(f"Token {token_mint} failed safety check: {safety.reason}")
                    return False
                    
                # Check if the token has sufficient liquidity
                if safety.liquidity_sol < float(os.getenv("MIN_EXIT_LIQUIDITY_SOL", "5.0")):
                    logger.warning(f"Insufficient liquidity for {token_mint}: {safety.liquidity_sol} SOL")
                    return False
            
            # Check if victim's trade is large enough to be worth front-running
            min_victim_size = float(os.getenv("SANDWICH_MIN_VICTIM_SIZE", "5.0"))
            max_victim_size = float(os.getenv("SANDWICH_MAX_VICTIM_SIZE", "320.0"))
            
            if victim_amount < min_victim_size:
                logger.debug(f"Victim amount too small: {victim_amount} SOL < {min_victim_size} SOL")
                return False
                
            if victim_amount > max_victim_size:
                logger.debug(f"Victim amount too large: {victim_amount} SOL > {max_victim_size} SOL")
                return False
            
            # Check if current balance allows this trade
            if self.balance_monitor:
                sol_balance = self.balance_monitor.sol_balance
                min_sol_balance = float(os.getenv("MIN_SOL_BALANCE", "0.03"))
                
                if sol_balance < FRONT_RUN_AMOUNT + min_sol_balance:
                    logger.warning(f"Insufficient SOL balance: {sol_balance} SOL < {FRONT_RUN_AMOUNT + min_sol_balance} SOL")
                    return False
            
            # In a production implementation, simulate the transaction to calculate exact profit
            # For now, we'll use a simple heuristic based on victim's trade size
            estimated_profit = victim_amount * 0.005  # 0.5% of victim's trade as estimated profit
            
            if estimated_profit < MIN_PROFIT:
                logger.debug(f"Estimated profit too low: {estimated_profit:.6f} SOL < {MIN_PROFIT} SOL")
                return False
                
            logger.info(f"Profitable opportunity detected for {token_mint}")
            logger.info(f"Victim amount: {victim_amount} SOL, Estimated profit: {estimated_profit:.6f} SOL")
            return True
            
        except Exception as e:
            logger.error(f"Error checking profitability: {e}")
            return False

    async def execute_front_run(self, swap_data: Dict, victim_signature: str):
        """Execute front-running strategy with GPU-optimized calculations"""
        if self.current_position or victim_signature in self.active_trades:
            return

        token_mint = swap_data.get("token_mint")
        if not token_mint:
            logger.error("Missing token mint in swap data")
            return
            
        self.active_trades.add(victim_signature)
        
        try:
            logger.info(f"Executing front-run for {token_mint}")
            logger.info(f"Victim transaction: {victim_signature}")
            
            # Generate correct instruction data for Raydium V4 AMM
            instruction_data = await asyncio.to_thread(
                self.trading_core.create_amm_instruction_data,
                FRONT_RUN_AMOUNT,  # amount_sol
                None  # min_amount_out (None = 0)
            )
            
            # Record start time for performance tracking
            start_time = time.time()
            
            # Execute the front-running swap
            tx_status = await self.trading_core.execute_custom_swap(
                token_mint=token_mint,
                instruction_data=instruction_data
            )

            # Calculate execution time
            execution_time = time.time() - start_time
            logger.info(f"Transaction execution time: {execution_time:.4f} seconds")

            if tx_status.success:
                logger.info(f"Front-run successful: {tx_status.signature}")
                self.stats['success_count'] += 1
                
                # Create position tracking data
                self.current_position = {
                    "entry_sig": tx_status.signature,
                    "victim_sig": victim_signature,
                    "token_mint": token_mint,
                    "entry_time": time.time(),
                    "entry_sol": FRONT_RUN_AMOUNT
                }
                
                # Schedule exit after a short delay to capture profit
                asyncio.create_task(self.schedule_exit())
            else:
                logger.error(f"Front-run failed: {tx_status.error}")
                self.stats['failure_count'] += 1

        except Exception as e:
            logger.error(f"Execution error: {e}")
            logger.error(traceback.format_exc())
            self.stats['failure_count'] += 1
        finally:
            self.active_trades.discard(victim_signature)

    async def schedule_exit(self):
        """Schedule position exit after delay with optimized timing"""
        try:
            # Get configurable exit delay
            delay = float(os.getenv("EXIT_DELAY", "3.0"))
            logger.info(f"Scheduling exit in {delay} seconds")
            
            # Wait for victim transaction to complete
            await asyncio.sleep(delay)
            
            # Attempt exit with retries
            max_attempts = 3
            for attempt in range(1, max_attempts + 1):
                exit_success = await self.exit_position()
                if exit_success:
                    break
                
                if attempt < max_attempts:
                    retry_delay = attempt * 2  # Exponential backoff
                    logger.warning(f"Exit attempt {attempt} failed, retrying in {retry_delay} seconds...")
                    await asyncio.sleep(retry_delay)
                    
            # If we still have a position after max attempts, keep trying in background
            if attempt == max_attempts and self.current_position:
                logger.warning("Failed to exit position after maximum attempts. " +
                              "Scheduling background retry...")
                asyncio.create_task(self.background_exit_retry())
                
        except Exception as e:
            logger.error(f"Error in schedule_exit: {e}")
            
    async def background_exit_retry(self):
        """Background task to keep trying position exit until successful"""
        retries = 0
        max_retries = 10
        
        while self.current_position and retries < max_retries and self.is_running:
            retries += 1
            delay = min(30, 5 * retries)  # Increasing delays with cap
            
            logger.info(f"Background exit retry #{retries} scheduled in {delay} seconds")
            await asyncio.sleep(delay)
            
            if not self.current_position:
                return  # Position already exited
                
            # Try to exit again
            success = await self.exit_position()
            if success:
                logger.info("Background exit retry succeeded")
                return
                
        if self.current_position:
            logger.error(f"Failed to exit position after {max_retries} background retries. " +
                          "Manual intervention may be required.")

    async def exit_position(self):
        """Exit current market position with dynamic pricing"""
        if not self.current_position:
            return True  # No position to exit

        try:
            token_mint = self.current_position["token_mint"]
            logger.info(f"Exiting position for {token_mint}")
            
            # Get token balance from monitor if available, otherwise fetch directly
            if self.balance_monitor and token_mint in self.balance_monitor.token_balances:
                token_balance = self.balance_monitor.token_balances.get(token_mint, 0)
            else:
                # Fetch balances directly
                balances = await self.trading_core.get_balances()
                _, token_balances = balances
                token_balance = token_balances.get(token_mint, 0)
            
            # Skip if balance is too small
            min_exit_balance = float(os.getenv("MIN_EXIT_BALANCE", "0.00001"))
            if token_balance <= min_exit_balance:
                logger.warning(f"Token balance too small to exit: {token_balance}")
                
                # Calculate profit based on SOL balance change
                profit = self.calculate_profit_from_balances()
                self.stats['last_profit'] = profit
                self.stats['profit_total'] += profit
                
                # Record position as closed
                self.current_position = None
                return True
                
            logger.info(f"Selling token balance: {token_balance}")
            
            # Calculate position duration
            entry_time = self.current_position.get("entry_time", 0)
            position_duration = time.time() - entry_time
            logger.info(f"Position held for {position_duration:.2f} seconds")
            
            # Create optimized instruction data for selling tokens
            # Using slippage buffer to ensure successful transaction
            slippage_buffer = 0.98  # 2% buffer
            exit_amount = token_balance * slippage_buffer
            
            instruction_data = await asyncio.to_thread(
                self.trading_core.create_amm_instruction_data,
                exit_amount,  # Use 98% of balance
                None  # min_amount_out (None = 0)
            )

            # Execute the exit swap with higher priority fee
            tx_status = await self.trading_core.execute_custom_swap(
                token_mint=token_mint,
                instruction_data=instruction_data
            )

            if tx_status.success:
                logger.info(f"Exit successful: {tx_status.signature}")
                
                # Calculate profit
                entry_sig = self.current_position.get("entry_sig", "")
                profit = self.calculate_profit_from_balances()
                
                logger.info(f"Trade complete: {entry_sig} -> {tx_status.signature}")
                logger.info(f"Estimated profit: {profit:.6f} SOL")
                
                # Update stats
                self.stats['last_profit'] = profit
                self.stats['profit_total'] += profit
                
                # Clear current position
                self.current_position = None
                return True
            else:
                logger.error(f"Exit failed: {tx_status.error}")
                return False

        except Exception as e:
            logger.error(f"Exit error: {e}")
            logger.error(traceback.format_exc())
            return False
            
    def calculate_profit_from_balances(self) -> float:
        """Calculate profit from balance changes"""
        if not self.balance_monitor or not self.current_position:
            return 0.0
            
        # Get entry amount
        entry_amount = self.current_position.get("entry_sol", 0.0)
        
        # Use balance monitor history to calculate profit
        profit = self.balance_monitor.calculate_profit(60.0)
        
        # Adjust for gas costs (estimate)
        estimated_gas = 0.001  # 0.001 SOL for gas
        
        return profit - estimated_gas
    
    async def run(self):
        """Main bot execution loop with proper cleanup"""
        logger.info("Starting Raydium MEV Bot with GPU acceleration")
        
        # Windows-compatible signal handling - remove signal handlers
        # (Windows does not support signal handling with asyncio)
        
        try:
            # Initialize the bot
            await self.initialize()
            
            # Start monitoring the mempool
            await self.monitor_mempool()
            
        except Exception as e:
            logger.error(f"Fatal error: {e}")
            logger.error(traceback.format_exc())
        finally:
            # Ensure proper cleanup
            await self.shutdown(None)
            
    async def shutdown(self, signal):
        """Handle shutdown with proper cleanup"""
        if not self.is_running:
            return  # Already shutting down
            
        self.is_running = False
        
        if signal:
            logger.info(f"Received exit signal, shutting down...")
        else:
            logger.info("Shutting down...")
        
        # Exit any current position before shutting down
        if self.current_position:
            logger.info("Exiting current position before shutdown")
            await self.exit_position()
            
        # Stop balance monitor
        if self.balance_monitor:
            logger.info("Stopping balance monitor")
            await self.balance_monitor.stop()
            
        # Release GPU resources
        if torch.cuda.is_available():
            logger.info("Releasing GPU resources")
            torch.cuda.empty_cache()
            
        logger.info("Shutdown complete")
        
        # Set shutdown event
        self.shutdown_event.set()
        
class nullcontext:
    """Context manager that does nothing, for conditional with statements"""
    def __enter__(self): pass
    def __exit__(self, *exc_info): pass

async def main():
    """Main entry point with proper startup and shutdown handling"""
    bot = RaydiumMEVBot()
    
    try:
        # Run the bot
        await bot.run()
        
        # Wait for shutdown signal
        await bot.shutdown_event.wait()
        
    except KeyboardInterrupt:
        logger.info("Bot stopped by user")
        await bot.shutdown(None)
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        logger.error(traceback.format_exc())
        await bot.shutdown(None)
        sys.exit(1)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info("Bot stopped by user")
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        logger.error(traceback.format_exc())
        sys.exit(1) please my eyes are bad i need to fix this script and have it fully written with the websocket part fixed