MEV Bot Detection with Ollama: Maximum Extractable Value Analysis Guide

# Install Ollama on macOS/Linux
curl -fsSL https://ollama.ai/install.sh | sh

# Start Ollama service
ollama serve

# Pull required models for analysis
ollama pull llama2
ollama pull codellama

# requirements.txt
web3==6.10.0
pandas==2.0.3
requests==2.31.0
numpy==1.24.3
matplotlib==3.7.2
ollama==0.1.7

pip install -r requirements.txt

import ollama
import json
from web3 import Web3
import pandas as pd
from datetime import datetime
import requests

class MEVBotDetector:
    def __init__(self, rpc_url, ollama_model="llama2"):
        """Initialize MEV bot detection system with Ollama AI"""
        self.w3 = Web3(Web3.HTTPProvider(rpc_url))
        self.ollama_model = ollama_model
        self.mev_patterns = {
            'sandwich_attack': [],
            'arbitrage': [],
            'liquidation': [],
            'front_running': []
        }
    
    def analyze_block_transactions(self, block_number):
        """Analyze all transactions in a block for MEV patterns"""
        block = self.w3.eth.get_block(block_number, full_transactions=True)
        transactions = []
        
        for tx in block.transactions:
            tx_data = {
                'hash': tx.hash.hex(),
                'from': tx['from'],
                'to': tx['to'],
                'value': tx['value'],
                'gas_price': tx['gasPrice'],
                'gas_limit': tx['gas'],
                'transaction_index': tx['transactionIndex']
            }
            transactions.append(tx_data)
        
        return self.detect_mev_patterns(transactions, block_number)
    
    def detect_mev_patterns(self, transactions, block_number):
        """Use Ollama to detect MEV patterns in transaction data"""
        # Prepare transaction data for analysis
        tx_summary = self.prepare_transaction_summary(transactions)
        
        # Create analysis prompt for Ollama
        prompt = f"""
        Analyze these blockchain transactions from block {block_number} for MEV bot activity:

        {tx_summary}

        Look for these patterns:
        1. Sandwich attacks: Same address before/after target transaction
        2. Arbitrage: Quick buy/sell across different exchanges
        3. Front-running: Similar transactions with higher gas prices
        4. Liquidations: Large position closures in lending protocols

        Respond with JSON format:
        {{
            "mev_detected": boolean,
            "pattern_type": "sandwich_attack|arbitrage|front_running|liquidation|none",
            "suspicious_addresses": ["address1", "address2"],
            "confidence_score": 0.0-1.0,
            "explanation": "detailed explanation"
        }}
        """
        
        # Query Ollama for analysis
        response = ollama.generate(
            model=self.ollama_model,
            prompt=prompt,
            format='json'
        )
        
        return json.loads(response['response'])
    
    def prepare_transaction_summary(self, transactions):
        """Format transaction data for Ollama analysis"""
        summary = []
        for i, tx in enumerate(transactions):
            gas_price_gwei = tx['gas_price'] / 1e9
            value_eth = tx['value'] / 1e18
            
            summary.append(f"TX{i}: From {tx['from'][:10]}... "
                         f"Value: {value_eth:.4f} ETH "
                         f"Gas: {gas_price_gwei:.1f} Gwei "
                         f"Index: {tx['transaction_index']}")
        
        return "\n".join(summary[:20])  # Limit for token efficiency

def detect_sandwich_attacks(self, transactions):
    """Specialized detection for sandwich attacks"""
    sandwich_candidates = []
    
    # Group transactions by target patterns
    for i in range(len(transactions) - 2):
        current = transactions[i]
        target = transactions[i + 1]
        following = transactions[i + 2]
        
        # Check if same address surrounds target transaction
        if (current['from'] == following['from'] and
            current['from'] != target['from'] and
            current['gas_price'] >= target['gas_price'] and
            following['gas_price'] <= target['gas_price']):
            
            sandwich_data = {
                'attacker': current['from'],
                'victim': target['from'],
                'front_tx': current['hash'],
                'target_tx': target['hash'],
                'back_tx': following['hash'],
                'profit_potential': self.calculate_sandwich_profit(current, target, following)
            }
            
            # Use Ollama to verify pattern
            if self.verify_sandwich_with_ollama(sandwich_data):
                sandwich_candidates.append(sandwich_data)
    
    return sandwich_candidates

def verify_sandwich_with_ollama(self, sandwich_data):
    """Use Ollama to verify sandwich attack pattern"""
    prompt = f"""
    Verify if this transaction pattern represents a sandwich attack:
    
    Attacker: {sandwich_data['attacker']}
    Victim: {sandwich_data['victim']}
    Profit Potential: {sandwich_data['profit_potential']} ETH
    
    Front transaction gas price vs target transaction gas price indicates priority.
    Back transaction completes the profit extraction.
    
    Is this a valid sandwich attack? Respond with true/false and confidence level.
    """
    
    response = ollama.generate(
        model=self.ollama_model,
        prompt=prompt
    )
    
    return 'true' in response['response'].lower()

import asyncio
import time

class RealTimeMEVMonitor:
    def __init__(self, detector, monitoring_interval=12):
        """Initialize real-time MEV monitoring"""
        self.detector = detector
        self.monitoring_interval = monitoring_interval
        self.alerts = []
        
    async def start_monitoring(self):
        """Start continuous MEV monitoring"""
        print("Starting real-time MEV monitoring...")
        
        while True:
            try:
                latest_block = self.detector.w3.eth.block_number
                
                # Analyze recent blocks
                for block_num in range(latest_block - 2, latest_block + 1):
                    results = self.detector.analyze_block_transactions(block_num)
                    
                    if results['mev_detected']:
                        await self.handle_mev_detection(block_num, results)
                
                # Wait for next block
                await asyncio.sleep(self.monitoring_interval)
                
            except Exception as e:
                print(f"Monitoring error: {e}")
                await asyncio.sleep(5)
    
    async def handle_mev_detection(self, block_number, detection_results):
        """Process MEV detection alerts"""
        alert = {
            'timestamp': datetime.now(),
            'block_number': block_number,
            'pattern_type': detection_results['pattern_type'],
            'confidence': detection_results['confidence_score'],
            'addresses': detection_results['suspicious_addresses']
        }
        
        self.alerts.append(alert)
        
        # Send alert notification
        print(f"🚨 MEV DETECTED in block {block_number}")
        print(f"Pattern: {alert['pattern_type']}")
        print(f"Confidence: {alert['confidence']:.2f}")
        print(f"Suspicious addresses: {alert['addresses']}")
        
        # Optional: Save to database or send webhook
        await self.save_alert(alert)
    
    async def save_alert(self, alert):
        """Save MEV detection alert"""
        # Implement database storage or external notification
        with open('mev_alerts.json', 'a') as f:
            json.dump(alert, f, default=str)
            f.write('\n')

# Initialize detector
detector = MEVBotDetector(
    rpc_url="https://eth-mainnet.alchemyapi.io/v2/YOUR_KEY",
    ollama_model="llama2"
)

# Analyze recent block for MEV activity
latest_block = detector.w3.eth.block_number
results = detector.analyze_block_transactions(latest_block)

print(f"MEV Analysis Results for Block {latest_block}:")
print(f"MEV Detected: {results['mev_detected']}")
print(f"Pattern Type: {results['pattern_type']}")
print(f"Confidence: {results['confidence_score']:.2f}")
print(f"Explanation: {results['explanation']}")

def analyze_historical_mev(detector, start_block, end_block):
    """Analyze MEV patterns across block range"""
    mev_summary = {
        'total_blocks': end_block - start_block,
        'mev_blocks': 0,
        'pattern_distribution': {},
        'top_addresses': {}
    }
    
    for block_num in range(start_block, end_block):
        results = detector.analyze_block_transactions(block_num)
        
        if results['mev_detected']:
            mev_summary['mev_blocks'] += 1
            pattern = results['pattern_type']
            
            # Track pattern distribution
            mev_summary['pattern_distribution'][pattern] = \
                mev_summary['pattern_distribution'].get(pattern, 0) + 1
            
            # Track suspicious addresses
            for addr in results['suspicious_addresses']:
                mev_summary['top_addresses'][addr] = \
                    mev_summary['top_addresses'].get(addr, 0) + 1
    
    return mev_summary

# Example: Analyze last 100 blocks
current_block = detector.w3.eth.block_number
historical_data = analyze_historical_mev(
    detector, 
    current_block - 100, 
    current_block
)

print(f"Historical MEV Analysis:")
print(f"MEV detected in {historical_data['mev_blocks']}/100 blocks")
print(f"Pattern distribution: {historical_data['pattern_distribution']}")

# Optimized batch processing
def batch_analyze_blocks(detector, block_range, batch_size=5):
    """Analyze blocks in batches for efficiency"""
    results = []
    
    for i in range(0, len(block_range), batch_size):
        batch = block_range[i:i + batch_size]
        batch_results = []
        
        for block_num in batch:
            result = detector.analyze_block_transactions(block_num)
            batch_results.append(result)
        
        results.extend(batch_results)
        time.sleep(1)  # Rate limiting
    
    return results

FROM python:3.9-slim

# Install Ollama
RUN curl -fsSL https://ollama.ai/install.sh | sh

# Copy application code
COPY . /app
WORKDIR /app

# Install dependencies
RUN pip install -r requirements.txt

# Pull required models
RUN ollama pull llama2

# Start monitoring
CMD ["python", "mev_monitor.py"]

# .env file
ETH_RPC_URL=https://eth-mainnet.alchemyapi.io/v2/YOUR_KEY
OLLAMA_MODEL=llama2
MONITORING_INTERVAL=12
ALERT_WEBHOOK_URL=https://hooks.slack.com/your-webhook

def test_ollama_connection():
    """Test Ollama service availability"""
    try:
        response = ollama.generate(model="llama2", prompt="Test connection")
        print("✅ Ollama connection successful")
        return True
    except Exception as e:
        print(f"❌ Ollama connection failed: {e}")
        return False

import time
from functools import wraps

def rate_limit(calls_per_second=1):
    """Rate limiting decorator for API calls"""
    def decorator(func):
        last_called = [0.0]
        
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = 1.0 / calls_per_second - elapsed
            
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        
        return wrapper
    return decorator

# Apply rate limiting to detection methods
@rate_limit(calls_per_second=2)
def rate_limited_analysis(self, block_number):
    return self.analyze_block_transactions(block_number)

from sklearn.ensemble import RandomForestClassifier
import numpy as np

class HybridMEVDetector(MEVBotDetector):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.ml_model = RandomForestClassifier(n_estimators=100)
        self.is_trained = False
    
    def extract_features(self, transactions):
        """Extract numerical features for ML model"""
        features = []
        
        for tx in transactions:
            feature_vector = [
                tx['gas_price'] / 1e9,  # Gas price in Gwei
                tx['value'] / 1e18,     # Value in ETH
                tx['gas_limit'],        # Gas limit
                tx['transaction_index'] # Position in block
            ]
            features.append(feature_vector)
        
        return np.array(features)
    
    def hybrid_detection(self, transactions, block_number):
        """Combine Ollama and ML analysis"""
        # Get Ollama analysis
        ollama_result = self.detect_mev_patterns(transactions, block_number)
        
        # Get ML prediction if model is trained
        if self.is_trained:
            features = self.extract_features(transactions)
            ml_prediction = self.ml_model.predict_proba(features)
            
            # Combine predictions
            combined_confidence = (
                ollama_result['confidence_score'] * 0.7 +
                np.mean(ml_prediction[:, 1]) * 0.3
            )
            
            ollama_result['confidence_score'] = combined_confidence
        
        return ollama_result