NFT MEV: Floor Price Manipulation and Sandwich Attacks

Primary NFT MEV Vectors:

# Floor Price Manipulation Detection System
import asyncio
import websockets
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import numpy as np
from collections import deque

@dataclass
class TradeEvent:
    timestamp: datetime
    collection_address: str
    token_id: str
    price: float
    seller_address: str
    buyer_address: str
    marketplace: str

class FloorPriceManipulationDetector:
    def __init__(self):
        # Window sizes for pattern detection
        self.price_window = deque(maxlen=100)  # Last 100 trades
        self.time_window = deque(maxlen=50)    # Last 50 trades
        self.volume_threshold = 5  # ETH
        self.time_threshold = 300  # 5 minutes
        
        # Manipulation pattern thresholds
        self.floor_drop_threshold = 0.15  # 15% drop triggers analysis
        self.volume_concentration = 0.7   # 70% from single address
        
    async def monitor_marketplace_stream(self, marketplace_url: str):
        """Monitor real-time NFT marketplace for manipulation patterns"""
        async with websockets.connect(marketplace_url) as websocket:
            while True:
                message = await websocket.recv()
                trade_event = self.parse_trade_event(message)
                
                if self.detect_floor_manipulation(trade_event):
                    await self.trigger_alert(trade_event)
                
                self.update_price_history(trade_event)
    
    def detect_floor_manipulation(self, trade: TradeEvent) -> bool:
        """Detect potential floor price manipulation patterns"""
        
        # Pattern 1: Rapid sequence of low-price sales
        recent_trades = [t for t in self.price_window 
                        if (trade.timestamp - t.timestamp).seconds < self.time_threshold]
        
        if len(recent_trades) >= 3:
            prices = [t.price for t in recent_trades]
            avg_price = np.mean(prices)
            
            # Check for coordinated low-price sales
            if (trade.price < avg_price * 0.8 and 
                all(t.price < avg_price * 0.85 for t in recent_trades[-3:])):
                return True
        
        # Pattern 2: High-volume single address trading
        address_trades = [t for t in self.price_window 
                         if t.seller_address == trade.seller_address]
        
        if len(address_trades) >= 5:
            total_volume = sum(t.price for t in address_trades)
            collection_volume = sum(t.price for t in self.price_window)
            
            if total_volume / collection_volume > self.volume_concentration:
                return True
        
        # Pattern 3: Sandwich pattern (buy high, sell low)
        if self.detect_sandwich_pattern(trade):
            return True
            
        return False
    
    def detect_sandwich_pattern(self, trade: TradeEvent) -> bool:
        """Detect sandwich attack patterns in NFT trading"""
        
        # Look for buy-then-sell sequences with price manipulation
        recent_trades = list(self.price_window)
        
        for i in range(len(recent_trades) - 1):
            if (recent_trades[i].buyer_address == trade.seller_address and
                recent_trades[i].collection_address == trade.collection_address):
                
                # Check for price difference suggesting manipulation
                buy_price = recent_trades[i].price
                sell_price = trade.price
                
                # Sell price significantly lower than buy price ( manipulation)
                if sell_price < buy_price * 0.7:
                    return True
        
        return False