USER
import asyncio
import json
import base64
import base58
from datetime import datetime
import websockets
import torch
import os
from typing import Optional
from dataclasses import dataclass
from solana_logger import logger
from config_loader import config
from solana_rust_bot import TradingCore, TradeParams
from solana_cli import SolanaCLI
@dataclass
class TradeOpportunity:
victim_data: bytes
amount: float
token_a_mint: str
token_b_mint: str
net_profit: float
instruction_type: int
transaction_id: str
class MEVBot:
def __init__(self):
# Load configuration
self.load_config()
# Initialize core components
self.init_components()
# Trading parameters from config
trade_config = config.get_trade_config()
self.POSITION_SIZE = trade_config['front_run_amount']
self.MIN_VICTIM_SIZE = trade_config['min_victim_size']
self.MAX_VICTIM_SIZE = trade_config['max_victim_size']
self.MIN_PROFIT = trade_config['min_profit']
self.LAMPORTS_PER_SOL = 1_000_000_000
# Track processed transactions
self.processed_txs = set()
# Balance monitoring
balance_config = config.get_balance_config()
self.min_sol_balance = balance_config['min_sol_balance']
self.min_wsol_balance = balance_config['min_wsol_balance']
self.balances = {"SOL": 0.0, "WSOL": 0.0}
self.is_trading_paused = False
# Performance tracking
self.active = True
self.trades_found = 0
self.trades_executed = 0
self.successful_trades = 0
self.total_profit = 0.0
self.start_time = datetime.now()
# Concurrency control
self.trade_lock = asyncio.Lock()
self.MAX_ACTIVE_TRADES = 1
self.active_trades = 0
# Setup CUDA
self._setup_cuda()
self.log_startup()
def load_config(self):
"""Load configuration and network settings"""
network_config = config.get_network_config()
self.ws_url = network_config['ws_url']
self.api_key = network_config['api_key']
self.program_id = config.get('raydium')
# Account configuration
self.wallet = config.get('wallet')
self.wsol_account = config.get('wsol_token_account')
# Market configuration
market_accounts = config.get_market_accounts('SOL-WSOL')
self.token_a_mint = market_accounts['token_a_mint']
self.token_b_mint = market_accounts['token_b_mint']
def init_components(self):
"""Initialize core components"""
try:
# Initialize trading core with RPC URL from config
self.trading_core = TradingCore(config.get_network_config()['rpc_url'])
# Initialize Solana CLI helper
self.solana_cli = SolanaCLI()
logger.system("Core components initialized successfully")
except Exception as e:
raise RuntimeError(f"Failed to initialize components: {e}")
def _setup_cuda(self):
"""Configure CUDA for optimal performance"""
if not torch.cuda.is_available():
raise RuntimeError("CUDA required!")
torch.cuda.empty_cache()
torch.set_float32_matmul_precision('high')
torch.backends.cudnn.benchmark = True
torch.backends.cudnn.enabled = True
self.cuda_stream = torch.cuda.Stream(priority=-1)
logger.system(f"Using GPU: {torch.cuda.get_device_name(0)}")
def log_startup(self):
"""Log startup information"""
logger.system(f"""
🚀 MEV Bot Starting
====================
Position Size: {self.POSITION_SIZE} SOL
Trade Range: {self.MIN_VICTIM_SIZE}-{self.MAX_VICTIM_SIZE} SOL
Min Profit: {self.MIN_PROFIT} SOL
====================
Wallet: {self.wallet}
WSOL Account: {self.wsol_account}
====================
""")
def calculate_profit(self, amount: float) -> float:
"""Calculate potential trade profit using CUDA"""
with torch.cuda.stream(self.cuda_stream):
amount_tensor = torch.tensor(amount, device='cuda')
if amount >= 40:
profit_rate = 0.025
elif amount >= 20:
profit_rate = 0.020
elif amount >= 5:
profit_rate = 0.015
else:
return 0.0
entry_impact = 0.001 * amount_tensor
exit_impact = entry_impact
priority_fee = 0.002
net_profit = (amount_tensor * profit_rate) - (entry_impact + exit_impact + priority_fee)
return float(net_profit) if net_profit > self.MIN_PROFIT else 0.0
def decode_trade_data(self, raw_data: bytes, tx_id: str) -> Optional[TradeOpportunity]:
"""Decode and validate trade data"""
try:
if tx_id in self.processed_txs:
return None
if len(raw_data) < 40:
return None
instruction = raw_data[0]
amount = int.from_bytes(raw_data[1:9], "little") / 1e9
if instruction != 0x03 or not (self.MIN_VICTIM_SIZE <= amount <= self.MAX_VICTIM_SIZE):
return None
token_a_bytes = raw_data[40:72] if len(raw_data) >= 72 else raw_data[-32:]
token_b_bytes = raw_data[72:104] if len(raw_data) >= 104 else None
token_a_mint = base58.b58encode(token_a_bytes).decode()
token_b_mint = base58.b58encode(token_b_bytes).decode() if token_b_bytes else "So11111111111111111111111111111111111111112"
profit = self.calculate_profit(amount)
if profit <= 0:
return None
self.processed_txs.add(tx_id)
return TradeOpportunity(
victim_data=raw_data,
amount=amount,
token_a_mint=token_a_mint,
token_b_mint=token_b_mint,
net_profit=profit,
instruction_type=instruction,
transaction_id=tx_id
)
except Exception as e:
logger.error(f"Trade data decode error: {e}")
return None
async def execute_trade(self, trade: TradeOpportunity) -> bool:
"""Execute trade with atomic entry"""
if self.is_trading_paused:
logger.warning("Trading is paused. Skipping trade execution.")
return False
async with self.trade_lock:
try:
if self.active_trades >= self.MAX_ACTIVE_TRADES:
logger.warning("Max active trades reached. Skipping trade execution.")
return False
self.active_trades += 1
position_lamports = int(self.POSITION_SIZE * self.LAMPORTS_PER_SOL)
logger.trade(f"""
🎯 Trade Found:
Size: {trade.amount:.4f} SOL
Position: {self.POSITION_SIZE} SOL ({position_lamports} lamports)
Token A: {trade.token_a_mint}
WSOL: {self.wsol_account}
Est. Profit: {trade.net_profit:.6f} SOL
TX ID: {trade.transaction_id}
""")
# Get market accounts in Raydium's required order
market_accounts = config.get_market_accounts('SOL-WSOL')
# Must be in this exact order for Raydium
raydium_accounts = [
config.get('raydium'), # Program ID
market_accounts['amm_open_orders'], # AMM Open Orders Account
market_accounts['amm_target'], # Pool Temp LP Token Account
market_accounts['market'], # Market ID
market_accounts['base_vault'], # Base Vault
market_accounts['quote_vault'], # Quote Vault
market_accounts['serum_market'], # Serum Market ID
market_accounts['serum_bids'], # Serum Bids
market_accounts['serum_asks'], # Serum Asks
market_accounts['event_queue'], # Event Queue
market_accounts['vault_signer'], # Vault Signer
self.wsol_account, # User Source Token Account (needs to be signer)
config.get('token'), # Token Program
self.wallet, # User Owner/Authority
]
# Execute entry trade using victim's instruction data
entry_params = TradeParams(
accounts=raydium_accounts,
amount_in=position_lamports,
is_buy=True,
reuse_blockhash=False,
token_a_mint=trade.token_a_mint,
token_b_mint=trade.token_b_mint,
victim_data=trade.victim_data,
sell_all=False
)
entry_result = await self.trading_core.execute_trade_py(entry_params)
if entry_result:
await asyncio.sleep(0.0001)
# Execute exit trade with same account structure
exit_params = TradeParams(
accounts=raydium_accounts,
amount_in=None, # Will use sell_all
is_buy=False,
reuse_blockhash=True,
token_a_mint=trade.token_a_mint,
token_b_mint=trade.token_b_mint,
victim_data=None,
sell_all=True
)
exit_result = await self.trading_core.execute_trade_py(exit_params)
if exit_result:
self.trades_executed += 1
self.successful_trades += 1
self.total_profit += trade.net_profit
logger.trade(f"""
💰 Trade Completed:
Entry TX: {entry_result}
Exit TX: {exit_result}
Original TX: {trade.transaction_id}
WSOL Account: {self.wsol_account}
Profit: {trade.net_profit:.6f} SOL
Total Profit: {self.total_profit:.6f} SOL
Success Rate: {(self.successful_trades/self.trades_executed*100):.1f}%
""")
return True
return False
except Exception as e:
logger.error(f"Trade execution error: {e}")
return False
finally:
self.active_trades -= 1
async def monitor_balances(self):
"""Monitor SOL and WSOL balances using WebSocket with subscription tracking"""
while self.active:
try:
async with websockets.connect(
self.ws_url,
extra_headers={'api-key': self.api_key}
) as ws:
accounts = {
1: ("SOL", self.wallet),
2: ("WSOL", self.wsol_account)
}
subs = {} # Track subscription IDs
# Send initial subscriptions
for sub_id, (acc_type, address) in accounts.items():
await ws.send(json.dumps({
"jsonrpc": "2.0",
"id": sub_id,
"method": "accountSubscribe",
"params": [
address,
{"encoding": "base64", "commitment": "confirmed"}
]
}))
# Process messages
async for msg in ws:
if not self.active:
break
data = json.loads(msg)
# Handle subscription responses
if "result" in data:
sub_id = data["id"]
if sub_id in accounts:
acc_type = accounts[sub_id][0]
subs[data["result"]] = acc_type
continue
# Handle balance updates
if "method" in data and data["method"] == "accountNotification":
sub_id = data["params"]["subscription"]
acc_type = subs.get(sub_id, "Unknown")
# Extract balance information
lamports = data["params"]["result"]["value"]["lamports"]
new_balance = lamports / self.LAMPORTS_PER_SOL
old_balance = self.balances.get(acc_type, 0)
# Update balances and log changes
if new_balance != old_balance:
self.balances[acc_type] = new_balance
change = new_balance - old_balance
logger.info(
f"{acc_type} Balance: {new_balance:.4f} SOL "
f"(Δ {change:+.4f} SOL)"
)
# Check balance thresholds
if (self.balances["SOL"] < self.min_sol_balance or
self.balances["WSOL"] < self.min_wsol_balance):
if not self.is_trading_paused:
logger.warning("Trading paused - Low balances:")
logger.warning(
f" SOL: {self.balances['SOL']:.4f}/"
f"{self.min_sol_balance:.4f} | "
f"WSOL: {self.balances['WSOL']:.4f}/"
f"{self.min_wsol_balance:.4f}"
)
self.is_trading_paused = True
else:
if self.is_trading_paused:
logger.info("Trading resumed - Balances restored")
self.is_trading_paused = False
except websockets.ConnectionClosed:
logger.warning("Balance monitor connection closed, reconnecting...")
await asyncio.sleep(1)
except Exception as e:
logger.error(f"Balance monitoring error: {e}")
await asyncio.sleep(1)
async def process_mempool(self):
"""Process Raydium mempool data"""
while self.active:
try:
async with websockets.connect(
self.ws_url,
extra_headers={
'api-key': self.api_key,
'User-Agent': 'solana-client/1.18.23'
},
ping_interval=5,
ping_timeout=10,
compression=None
) as ws:
logger.info("Mempool WebSocket Connected")
await ws.send(json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "logsSubscribe",
"params": [
{"mentions": [self.program_id]},
{"commitment": "processed"}
]
}))
await ws.recv() # Subscription confirmation
while self.active:
try:
msg = await ws.recv()
data = json.loads(msg)
if "params" in data:
result = data["params"].get("result", {})
logs = result.get("value", {}).get("logs", [])
tx_id = result.get("value", {}).get("signature", "unknown")
for log in logs:
if "ray_log:" in log:
raw_data = base64.b64decode(log.split("ray_log:")[1].strip())
trade = self.decode_trade_data(raw_data, tx_id)
if trade:
self.trades_found += 1
await self.execute_trade(trade)
except Exception as e:
logger.info(f"Message processing: {e}")
continue
except websockets.ConnectionClosed:
logger.warning("Mempool WebSocket disconnected, reconnecting...")
await asyncio.sleep(1)
except Exception as e:
logger.error(f"Mempool connection error: {e}")
await asyncio.sleep(1)
def show_performance(self):
"""Display trading performance metrics"""
runtime = (datetime.now() - self.start_time).total_seconds()
hours = int(runtime // 3600)
minutes = int((runtime % 3600) // 60)
seconds = int(runtime % 60)
logger.system(f"""
🏁 Trading Performance Report
============================
Runtime: {hours:02d}:{minutes:02d}:{seconds:02d}
Trade Statistics:
----------------
Opportunities Found: {self.trades_found}
Trades Executed: {self.trades_executed}
Successful Trades: {self.successful_trades}
Success Rate: {(self.successful_trades/max(1,self.trades_executed)*100):.1f}%
Profit Metrics:
--------------
Total Profit: {self.total_profit:.6f} SOL
Profit per Hour: {(self.total_profit/(runtime/3600)):.6f} SOL/hr
Average Profit per Trade: {(self.total_profit/max(1,self.successful_trades)):.6f} SOL
Current Balances:
----------------
SOL: {self.balances['SOL']:.4f}
WSOL: {self.balances['WSOL']:.4f}
============================
""")
async def run(self):
"""Start bot with all monitoring systems"""
try:
# Start monitoring systems
balance_task = asyncio.create_task(self.monitor_balances())
mempool_task = asyncio.create_task(self.process_mempool())
# Start balance manager
balance_manager_task = asyncio.create_task(self.solana_cli.start())
# Show initial balance check
await asyncio.sleep(2) # Wait for first balance update
logger.info(f"""
Initial Balance Check:
SOL: {self.balances['SOL']:.4f}
WSOL: {self.balances['WSOL']:.4f}
""")
# Wait for tasks
await asyncio.gather(balance_task, mempool_task, balance_manager_task)
except asyncio.CancelledError:
logger.system("Tasks cancelled")
except Exception as e:
logger.error(f"Task error: {e}")
finally:
self.show_performance()
await self.cleanup()
async def cleanup(self):
"""Cleanup resources"""
self.active = False
await self.solana_cli.cleanup()
if torch.cuda.is_available():
torch.cuda.empty_cache()
logger.system("Cleanup completed")
async def main():
if os.name == 'nt':
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
try:
bot = MEVBot()
await bot.run()
except KeyboardInterrupt:
logger.system("Bot stopping gracefully...")
except Exception as e:
logger.error(f"Fatal error: {e}")
finally:
# Cleanup
if torch.cuda.is_available():
torch.cuda.empty_cache()
logger.system("Bot stopped")
if __name__ == "__main__":
asyncio.run(main()) write the full script please word for word without the indentation issuesASSISTANT
Certainly! Here is the full script with all indentation issues resolved:
```python
import asyncio
import json
import base64
import base58
from datetime import datetime
import websockets
import torch
import os
from typing import Optional
from dataclasses import dataclass
from solana_logger import logger
from config_loader import config
from solana_rust_bot import TradingCore, TradeParams
from solana_cli import SolanaCLI
@dataclass
class TradeOpportunity:
victim_data: bytes
amount: float
token_a_mint: str
token_b_mint: str
net_profit: float
instruction_type: int
transaction_id: str
class MEVBot:
def __init__(self):
# Load configuration
self.load_config()
# Initialize core components
self.init_components()
# Trading parameters from config
trade_config = config.get_trade_config()
self.POSITION_SIZE = trade_config['front_run_amount']
self.MIN_VICTIM_SIZE = trade_config['min_victim_size']
self.MAX_VICTIM_SIZE = trade_config['max_victim_size']
self.MIN_PROFIT = trade_config['min_profit']
self.LAMPORTS_PER_SOL = 1_000_000_000
# Track processed transactions
self.processed_txs = set()
# Balance monitoring
balance_config = config.get_balance_config()
self.min_sol_balance = balance_config['min_sol_balance']
self.min_wsol_balance = balance_config['min_wsol_balance']
self.balances = {"SOL": 0.0, "WSOL": 0.0}
self.is_trading_paused = False
# Performance tracking
self.active = True
self.trades_found = 0
self.trades_executed = 0
self.successful_trades = 0
self.total_profit = 0.0
self.start_time = datetime.now()
# Concurrency control
self.trade_lock = asyncio.Lock()
self.MAX_ACTIVE_TRADES = 1
self.active_trades = 0
# Setup CUDA
self._setup_cuda()
self.log_startup()
def load_config(self):
"""Load configuration and network settings"""
network_config = config.get_network_config()
self.ws_url = network_config['ws_url']
self.api_key = network_config['api_key']
self.program_id = config.get('raydium')
# Account configuration
self.wallet = config.get('wallet')
self.wsol_account = config.get('wsol_token_account')
# Market configuration
market_accounts = config.get_market_accounts('SOL-WSOL')
self.token_a_mint = market_accounts['token_a_mint']
self.token_b_mint = market_accounts['token_b_mint']
def init_components(self):
"""Initialize core components"""
try:
# Initialize trading core with RPC URL from config
self.trading_core = TradingCore(config.get_network_config()['rpc_url'])
# Initialize Solana CLI helper
self.solana_cli = SolanaCLI()
logger.system("Core components initialized successfully")
except Exception as e:
raise RuntimeError(f"Failed to initialize components: {e}")
def _setup_cuda(self):
"""Configure CUDA for optimal performance"""
if not torch.cuda.is_available():
raise RuntimeError("CUDA required!")
torch.cuda.empty_cache()
torch.set_float32_matmul_precision('high')
torch.backends.cudnn.benchmark = True
torch.backends.cudnn.enabled = True
self.cuda_stream = torch.cuda.Stream(priority=-1)
logger.system(f"Using GPU: {torch.cuda.get_device_name(0)}")
def log_startup(self):
"""Log startup information"""
logger.system(f"""
🚀 MEV Bot Starting
====================
Position Size: {self.POSITION_SIZE} SOL
Trade Range: {self.MIN_VICTIM_SIZE}-{self.MAX_VICTIM_SIZE} SOL
Min Profit: {self.MIN_PROFIT} SOL
====================
Wallet: {self.wallet}
WSOL Account: {self.wsol_account}
====================
""")
def calculate_profit(self, amount: float) -> float:
"""Calculate potential trade profit using CUDA"""
with torch.cuda.stream(self.cuda_stream):
amount_tensor = torch.tensor(amount, device='cuda')
if amount >= 40:
profit_rate = 0.025
elif amount >= 20:
profit_rate = 0.020
elif amount >= 5:
profit_rate = 0.015
else:
return 0.0
entry_impact = 0.001 * amount_tensor
exit_impact = entry_impact
priority_fee = 0.002
net_profit = (amount_tensor * profit_rate) - (entry_impact + exit_impact + priority_fee)
return float(net_profit) if net_profit > self.MIN_PROFIT else 0.0
def decode_trade_data(self, raw_data: bytes, tx_id: str) -> Optional[TradeOpportunity]:
"""Decode and validate trade data"""
try:
if tx_id in self.processed_txs:
return None
if len(raw_data) < 40:
return None
instruction = raw_data[0]
amount = int.from_bytes(raw_data[1:9], "little") / 1e9
if instruction != 0x03 or not (self.MIN_VICTIM_SIZE <= amount <= self.MAX_VICTIM_SIZE):
return None
token_a_bytes = raw_data[40:72] if len(raw_data) >= 72 else raw_data[-32:]
token_b_bytes = raw_data[72:104] if len(raw_data) >= 104 else None
token_a_mint = base58.b58encode(token_a_bytes).decode()
token_b_mint = base58.b58encode(token_b_bytes).decode() if token_b_bytes else "So11111111111111111111111111111111111111112"
profit = self.calculate_profit(amount)
if profit <= 0:
return None
self.processed_txs.add(tx_id)
return TradeOpportunity(
victim_data=raw_data,
amount=amount,
token_a_mint=token_a_mint,
token_b_mint=token_b_mint,
net_profit=profit,
instruction_type=instruction,
transaction_id=tx_id
)
except Exception as e:
logger.error(f"Trade data decode error: {e}")
return None
async def execute_trade(self, trade: TradeOpportunity) -> bool:
"""Execute trade with atomic entry"""
if self.is_trading_paused:
logger.warning("Trading is paused. Skipping trade execution.")
return False
async with self.trade_lock:
try:
if self.active_trades >= self.MAX_ACTIVE_TRADES:
logger.warning("Max active trades reached. Skipping trade execution.")
return False
self.active_trades += 1
position_lamports = int(self.POSITION_SIZE * self.LAMPORTS_PER_SOL)
logger.trade(f"""
🎯 Trade Found:
Size: {trade.amount:.4f} SOL
Position: {self.POSITION_SIZE} SOL ({position_lamports} lamports)
Token A: {trade.token_a_mint}
WSOL: {self.wsol_account}
Est. Profit: {trade.net_profit:.6f} SOL
TX ID: {trade.transaction_id}
""")
# Get market accounts in Raydium's required order
market_accounts = config.get_market_accounts('SOL-WSOL')
# Must be in this exact order for Raydium
raydium_accounts = [
config.get('raydium'), # Program ID
market_accounts['amm_open_orders'], # AMM Open Orders Account
market_accounts['amm_target'], # Pool Temp LP Token Account
market_accounts['market'], # Market ID
market_accounts['base_vault'], # Base Vault
market_accounts['quote_vault'], # Quote Vault
market_accounts['serum_market'], # Serum Market ID
market_accounts['serum_bids'], # Serum Bids
market_accounts['serum_asks'], # Serum Asks
market_accounts['event_queue'], # Event Queue
market_accounts['vault_signer'], # Vault Signer
self.wsol_account, # User Source Token Account (needs to be signer)
config.get('token'), # Token Program
self.wallet, # User Owner/Authority
]
# Execute entry trade using victim's instruction data
entry_params = TradeParams(
accounts=raydium_accounts,
amount_in=position_lamports,
is_buy=True,
reuse_blockhash=False,
token_a_mint=trade.token_a_mint,
token_b_mint=trade.token_b_mint,
victim_data=trade.victim_data,
sell_all=False
)
entry_result = await self.trading_core.execute_trade_py(entry_params)
if entry_result:
await asyncio.sleep(0.0001)
# Execute exit trade with same account structure
exit_params = TradeParams(
accounts=raydium_accounts,
amount_in=None, # Will use sell_all
is_buy=False,
reuse_blockhash=True,
token_a_mint=trade.token_a_mint,
token_b_mint=trade.token_b_mint,
victim_data=None,
sell_all=True
)
exit_result = await self.trading_core.execute_trade_py(exit_params)
if exit_result:
self.trades_executed += 1
self.successful_trades += 1
self.total_profit += trade.net_profit
logger.trade(f"""
💰 Trade Completed:
Entry TX: {entry_result}
Exit TX: {exit_result}
Original TX: {trade.transaction_id}
WSOL Account: {self.wsol_account}
Profit: {trade.net_profit:.6f} SOL
Total Profit: {self.total_profit:.6f} SOL
Success Rate: {(self.successful_trades/self.trades_executed*100):.1f}%
""")
return True
return False
except Exception as e:
logger.error(f"Trade execution error: {e}")
return False
finally:
self.active_trades -= 1
async def monitor_balances(self):
"""Monitor SOL and WSOL balances using WebSocket with subscription tracking"""
while self.active:
try:
async with websockets.connect(
self.ws_url,
extra_headers={'api-key': self.api_key}
) as ws:
accounts = {
1: ("SOL", self.wallet),
2: ("WSOL", self.wsol_account)
}
subs = {} # Track subscription IDs
# Send initial subscriptions
for sub_id, (acc_type, address) in accounts.items():
await ws.send(json.dumps({
"jsonrpc": "2.0",
"id": sub_id,
"method": "accountSubscribe",
"params": [
address,
{"encoding": "base64", "commitment": "confirmed"}
]
}))
# Process messages
async for msg in ws:
if not self.active:
break
data = json.loads(msg)
# Handle subscription responses
if "result" in data:
sub_id = data["id"]
if sub_id in accounts:
acc_type = accounts[sub_id][0]
subs[data["result"]] = acc_type
continue
# Handle balance updates
if "method" in data and data["method"] == "accountNotification":
sub_id = data["params"]["subscription"]
acc_type = subs.get(sub_id, "Unknown")
# Extract balance information
lamports = data["params"]["result"]["value"]["lamports"]
new_balance = lamports / self.LAMPORTS_PER_SOL
old_balance = self.balances.get(acc_type, 0)
# Update balances and log changes
if new_balance != old_balance:
self.balances[acc_type] = new_balance
change = new_balance - old_balance
logger.info(
f"{acc_type} Balance: {new_balance:.4f} SOL "
f"(Δ {change:+.4f} SOL)"
)
# Check balance thresholds
if (self.balances["SOL"] < self.min_sol_balance or
self.balances["WSOL"] < self.min_wsol_balance):
if not self.is_trading_paused:
logger.warning("Trading paused - Low balances:")
logger.warning(
f" SOL: {self.balances['SOL']:.4f}/"
f"{self.min_sol_balance:.4f} | "
f"WSOL: {self.balances['WSOL']:.4f}/"
f"{self.min_wsol_balance:.4f}"
)
self.is_trading_paused = True
else:
if self.is_trading_paused:
logger.info("Trading resumed - Balances restored")
self.is_trading_paused = False
except websockets.ConnectionClosed:
logger.warning("Balance monitor connection closed, reconnecting...")
await asyncio.sleep(1)
except Exception as e:
logger.error(f"Balance monitoring error: {e}")
await asyncio.sleep(1)
async def process_mempool(self):
"""Process Raydium mempool data"""
while self.active:
try:
async with websockets.connect(
self.ws_url,
extra_headers={
'api-key': self.api_key,
'User-Agent': 'solana-client/1.18.23'
},
ping_interval=5,
ping_timeout=10,
compression=None
) as ws:
logger.info("Mempool WebSocket Connected")
await ws.send(json.dumps({
"jsonrpc": "2.0",
"id": 1,
"method": "logsSubscribe",
"params": [
{"mentions": [self.program_id]},
{"commitment": "processed"}
]
}))
await ws.recv() # Subscription confirmation
while self.active:
try:
msg = await ws.recv()
data = json.loads(msg)
if "params" in data:
result = data["params"].get("result", {})
logs = result.get("value", {}).get("logs", [])
tx_id = result.get("value", {}).get("signature", "unknown")
for log in logs:
if "ray_log:" in log:
raw_data = base64.b64decode(log.split("ray_log:")[1].strip())
trade = self.decode_trade_data(raw_data, tx_id)
if trade:
self.trades_found += 1
await self.execute_trade(trade)
except Exception as e:
logger.info(f"Message processing: {e}")
continue
except websockets.ConnectionClosed:
logger.warning("Mempool WebSocket disconnected, reconnecting...")
await asyncio.sleep(1)
except Exception as e:
logger.error(f"Mempool connection error: {e}")
await asyncio.sleep(1)
def show_performance(self):
"""Display trading performance metrics"""
runtime = (datetime.now() - self.start_time).total_seconds()
hours = int(runtime // 3600)
minutes = int((runtime % 3600) // 60)
seconds = int(runtime % 60)
logger.system(f"""
🏁 Trading Performance Report
============================
Runtime: {hours:02d}:{minutes:02d}:{seconds:02d}
Trade Statistics:
----------------
Opportunities Found: {self.trades_found}
Trades Executed: {self.trades_executed}
Successful Trades: {self.successful_trades}
Success Rate: {(self.successful_trades/max(1,self.trades_executed)*100):.1f}%
Profit Metrics:
--------------
Total Profit: {self.total_profit:.6f} SOL
Profit per Hour: {(self.total_profit/(runtime/3600)):.6f} SOL/hr
Average Profit per Trade: {(self.total_profit/max(1,self.successful_trades)):.6f} SOL
Current Balances:
----------------
SOL: {self.balances['SOL']:.4f}
WSOL: {self.balances['WSOL']:.4f}
============================
""")
async def run(self):
"""Start bot with all monitoring systems"""
try:
# Start monitoring systems
balance_task = asyncio.create_task(self.monitor_balances())
mempool_task = asyncio.create_task(self.process_mempool())
# Start balance manager
balance_manager_task = asyncio.create_task(self.solana_cli.start())
# Show initial balance check
await asyncio.sleep(2) # Wait for first balance update
logger.info(f"""
Initial Balance Check:
SOL: {self.balances['SOL']:.4f}
WSOL: {self.balances['WSOL']:.4f}
""")
# Wait for tasks
await asyncio.gather(balance_task, mempool_task, balance_manager_task)
except asyncio.CancelledError:
logger.system("Tasks cancelled")
except Exception as e:
logger.error(f"Task error: {e}")
finally:
self.show_performance()
await self.cleanup()
async def cleanup(self):
"""Cleanup resources"""
self.active = False
await self.solana_cli.cleanup()
if torch.cuda.is_available():
torch.cuda.empty_cache()
logger.system("Cleanup completed")
async def main():
if os.name == 'nt':
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
try:
bot = MEVBot()
await bot.run()
except KeyboardInterrupt:
logger.system("Bot stopping gracefully...")
except Exception as e:
logger.error(f"Fatal error: {e}")
finally:
# Cleanup
if torch.cuda.is_available():
torch.cuda.empty_cache()
logger.system("Bot stopped")
if __name__ == "__main__":
asyncio.run(main())
```
This code maintains proper indentation and formatting, allowing for clear readability and functionality. If you have any further questions or need additional modifications, feel free to ask!