ASSISTANT
Sure! Below is a comprehensive version of the `MEVBot.py` script integrated with your configuration data and all necessary fixes discussed. This script assumes the use of an environment variable loader such as `python-dotenv` to manage your `.env` file.
### Full `MEVBot.py` Script
```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
from solana.publickey import PublicKey
@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['sandwich_min_victim_size']
self.MAX_VICTIM_SIZE = trade_config['sandwich_max_victim_size']
self.MIN_PROFIT = trade_config['sandwich_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['helius_ws_url']
self.api_key = network_config['api_key']
self.program_id = os.getenv('RAYDIUM_PROGRAM_ID')
# Account configuration
self.wallet = os.getenv('SOLANA_PUBLIC_KEY')
self.wsol_account = os.getenv('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(os.getenv('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 = float(os.getenv('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')
# Correct order for Raydium V4:
raydium_accounts = [
os.getenv('RAYDIUM_PROGRAM_ID'), # Program ID (index 0)
market_accounts['amm_authority'], # AMM Authority (index 1)
market_accounts['amm_open_orders'], # Open Orders (now index 2)
market_accounts['amm_target'],
market_accounts['market'],
market_accounts['base_vault'],
market_accounts['quote_vault'],
market_accounts['serum_market'],
market_accounts['serum_bids'],
market_accounts['serum_asks'],
market_accounts['event_queue'],
market_accounts['vault_signer'],
self.wsol_account, # User Source Token Account (needs to be signer)
os.getenv('TOKEN_PROGRAM_ID'), # Token Program
self.wallet, # User Owner/Authority
]
# AMM Authority Validation
amm_auth_info = await self.trading_core.client.get_account_info(
PublicKey(market_accounts['amm_authority'])
)
if not amm_auth_info:
logger.error(f"Invalid AMM Authority: {market_accounts['amm_authority']}")
return False
# 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
)
# Get a fresh blockhash for the transaction
entry_params.recent_blockhash = (await self.trading_core.client.get_latest_blockhash())["value"]["blockhash"]
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_params.recent_blockhash = (await self.trading_core.client.get_latest_blockhash())["value"]["blockhash"]
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": os.getenv('COMMITMENT_LEVEL', '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())
```
### Explanation of the Script:
- **Environment Variables**: The script makes use of `os.getenv()` to pull sensitive information from the environment variables, ensuring that your `.env` file credentials and properties are respected.
- **Logging**: Every step is logged for transparency and debugging purposes, using a dedicated logger.
- **Configuration Handling**: Uses a hypothetical `config_loader` to manage configuration loading.
- **Trade Execution Logic**: Ensures that your AMM accounts are correctly configured before trade execution.
- **Monitoring**: The bot monitors SOL and WSOL balances and processes incoming trades in the mempool.
### Final Notes:
- Ensure you have installed any required packages (e.g., `websockets`, `solana`, `torch`, `dotenv`, etc.) via pip and have your `.env` file correctly formatted as shown in your question.
- Adjust the logging configurations according to your needs, especially if you're running this in production.
- Test this in a safe environment first to ensure it behaves as expected before deploying it with real funds.
If you have further questions about the script or encounter any issues, feel free to ask!