Slippage Protection for AI Agents: MEV, Sandwiching, and Frontrunning

"""
Slippage-Protected Swap Agent with MEV detection.
Uses Purple Flea Trading API for MEV-resistant execution.
"""

import asyncio
import logging
import random
import time
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
import aiohttp
import statistics

log = logging.getLogger('swap_agent')

PURPLEFLEA_TRADING = "https://purpleflea.com/trading-api"
API_KEY = "your_api_key_here"

# MEV detection thresholds
MEMPOOL_SCAN_INTERVAL_S = 1.0
SANDWICH_DETECTION_WINDOW = 5      # seconds to monitor before submitting
GAS_SPIKE_MULTIPLIER     = 2.5     # gas spike above median → possible bot activity
FRONTRUN_PRICE_MOVE_PCT  = 0.002   # 0.2% mid-price move in last 30s → delay

# Swap settings
LARGE_ORDER_THRESHOLD_USD = 10_000  # split orders above this size
SPLIT_CHUNK_USD           = 3_000   # chunk size for split orders
SPLIT_DELAY_S             = 12      # ~1 block between chunks
MAX_SLIPPAGE_BPS          = 50      # 0.50% absolute max slippage
MIN_SLIPPAGE_BPS          = 10      # 0.10% absolute min slippage


@dataclass
class MarketSnapshot:
    pair: str
    mid_price: float
    bid: float
    ask: float
    pool_depth_usd: float
    recent_prices: List[float]
    avg_gas_gwei: float
    timestamp: float


@dataclass
class SwapResult:
    tx_hash: str
    pair: str
    input_amount: float
    output_amount: float
    effective_price: float
    slippage_bps: int
    mev_risk_at_submission: str
    timestamp: float


class MEVDetector:
    """Heuristics for detecting active MEV bot presence before a swap."""

    def __init__(self):
        self.price_history: Dict[str, List[Tuple[float, float]]] = {}  # pair → [(ts, price)]
        self.gas_history: List[float] = []

    def update(self, snapshot: MarketSnapshot):
        pair = snapshot.pair
        if pair not in self.price_history:
            self.price_history[pair] = []

        now = time.time()
        self.price_history[pair].append((now, snapshot.mid_price))
        self.gas_history.append(snapshot.avg_gas_gwei)

        # Keep only last 60 seconds
        cutoff = now - 60
        self.price_history[pair] = [(t, p) for t, p in self.price_history[pair] if t > cutoff]
        if len(self.gas_history) > 60:
            self.gas_history = self.gas_history[-60:]

    def mev_risk(self, pair: str, trade_size_usd: float) -> Tuple[str, str]:
        """
        Returns (risk_level, reason) where risk_level in {LOW, MEDIUM, HIGH}.
        """
        reasons = []

        # Check 1: Recent price volatility (possible frontrunning in progress)
        history = self.price_history.get(pair, [])
        if len(history) >= 3:
            recent_window = [(t, p) for t, p in history if time.time() - t < 30]
            if len(recent_window) >= 2:
                prices = [p for _, p in recent_window]
                price_move_pct = abs(prices[-1] - prices[0]) / prices[0]
                if price_move_pct > FRONTRUN_PRICE_MOVE_PCT:
                    reasons.append(f"Price moved {price_move_pct:.2%} in 30s")

        # Check 2: Gas spike (bots bidding up to frontrun)
        if len(self.gas_history) >= 5:
            median_gas = statistics.median(self.gas_history[:-3])
            current_gas = self.gas_history[-1]
            if current_gas > median_gas * GAS_SPIKE_MULTIPLIER:
                reasons.append(f"Gas spiked {current_gas:.1f} vs median {median_gas:.1f} gwei")

        # Check 3: Trade size relative to pool depth
        pool_depth = self.price_history.get(pair, [])
        impact_estimate = trade_size_usd / 1_000_000  # rough 0.1% per $10K in $10M pool
        if impact_estimate > 0.003:  # > 0.3% impact
            reasons.append(f"High price impact: ~{impact_estimate:.1%}")

        if len(reasons) >= 2:
            return "HIGH", "; ".join(reasons)
        elif len(reasons) == 1:
            return "MEDIUM", reasons[0]
        return "LOW", "normal conditions"


class SlippageCalculator:
    """Dynamic slippage tolerance based on pool conditions."""

    @staticmethod
    def optimal_slippage(
        trade_usd: float,
        pool_depth_usd: float,
        recent_volatility_pct: float,
        mev_risk: str,
    ) -> int:
        """Return optimal slippage tolerance in basis points."""
        # Base slippage from price impact
        impact_bps = int((trade_usd / pool_depth_usd) * 10_000 * 2)  # rough AMM formula

        # Volatility buffer: allow for price movement during block time
        volatility_bps = int(recent_volatility_pct * 10_000 * 2)

        base = max(MIN_SLIPPAGE_BPS, impact_bps + volatility_bps)

        # Reduce tolerance under high MEV conditions to limit extraction
        if mev_risk == "HIGH":
            base = min(base, MIN_SLIPPAGE_BPS + 10)  # tighten to ~20 bps under attack
        elif mev_risk == "MEDIUM":
            base = min(base, 30)

        return min(base, MAX_SLIPPAGE_BPS)


class ProtectedSwapAgent:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.detector = MEVDetector()
        self.calc = SlippageCalculator()
        self._session: Optional[aiohttp.ClientSession] = None

    async def __aenter__(self):
        self._session = aiohttp.ClientSession(
            headers={"Authorization": f"Bearer {self.api_key}"}
        )
        return self

    async def __aexit__(self, *args):
        await self._session.close()

    async def get_market(self, pair: str) -> MarketSnapshot:
        """Fetch current market data for MEV assessment."""
        async with self._session.get(
            f"{PURPLEFLEA_TRADING}/market/{pair}"
        ) as resp:
            resp.raise_for_status()
            d = await resp.json()
        return MarketSnapshot(
            pair=pair,
            mid_price=d["mid"],
            bid=d["bid"],
            ask=d["ask"],
            pool_depth_usd=d.get("pool_depth_usd", 5_000_000),
            recent_prices=d.get("recent_prices", []),
            avg_gas_gwei=d.get("avg_gas_gwei", 20),
            timestamp=time.time(),
        )

    async def _execute_single_swap(
        self,
        pair: str,
        side: str,
        amount_usd: float,
        slippage_bps: int,
        mev_risk: str,
    ) -> SwapResult:
        """Execute a single swap chunk via Purple Flea Trading API.
        The API routes through MEV-resistant channels automatically."""

        # Randomize timing slightly to avoid bot pattern detection (±2 seconds)
        jitter = random.uniform(-2, 2)
        if jitter > 0:
            await asyncio.sleep(jitter)

        payload = {
            "pair":         pair,
            "side":         side,
            "amount_usd":   amount_usd,
            "slippage_bps": slippage_bps,
            "mev_protect":  True,     # Enable Purple Flea MEV protection
            "deadline_s":   30,       # Expire if not included in 30s
        }

        async with self._session.post(
            f"{PURPLEFLEA_TRADING}/swap", json=payload
        ) as resp:
            resp.raise_for_status()
            data = await resp.json()

        return SwapResult(
            tx_hash=data["tx_hash"],
            pair=pair,
            input_amount=amount_usd,
            output_amount=float(data["output_amount"]),
            effective_price=float(data["effective_price"]),
            slippage_bps=int(data["actual_slippage_bps"]),
            mev_risk_at_submission=mev_risk,
            timestamp=time.time(),
        )

    async def swap(self, pair: str, side: str, amount_usd: float) -> List[SwapResult]:
        """
        Execute a fully protected swap with:
        - MEV risk assessment
        - Dynamic slippage calculation
        - Order splitting for large trades
        - Timing randomization
        """
        results = []

        # Pre-trade market scan
        snapshot = await self.get_market(pair)
        self.detector.update(snapshot)

        # Compute recent volatility from price history
        prices = snapshot.recent_prices
        recent_vol = 0.0
        if len(prices) >= 2:
            returns = [abs(prices[i] / prices[i-1] - 1) for i in range(1, len(prices))]
            recent_vol = statistics.mean(returns)

        mev_risk, mev_reason = self.detector.mev_risk(pair, amount_usd)
        log.info(f"MEV risk: {mev_risk} — {mev_reason}")

        # If HIGH risk, wait for conditions to improve (up to 2 minutes)
        if mev_risk == "HIGH":
            log.warning(f"HIGH MEV risk detected. Waiting up to 2 minutes...")
            for _ in range(12):
                await asyncio.sleep(10)
                snapshot = await self.get_market(pair)
                self.detector.update(snapshot)
                mev_risk, mev_reason = self.detector.mev_risk(pair, amount_usd)
                if mev_risk != "HIGH":
                    log.info(f"MEV risk reduced to {mev_risk}, proceeding")
                    break

        slippage_bps = self.calc.optimal_slippage(
            trade_usd=min(amount_usd, LARGE_ORDER_THRESHOLD_USD),
            pool_depth_usd=snapshot.pool_depth_usd,
            recent_volatility_pct=recent_vol,
            mev_risk=mev_risk,
        )
        log.info(f"Dynamic slippage: {slippage_bps} bps")

        # Split large orders
        if amount_usd > LARGE_ORDER_THRESHOLD_USD:
            chunks = []
            remaining = amount_usd
            while remaining > 0:
                chunk = min(remaining, SPLIT_CHUNK_USD)
                chunks.append(chunk)
                remaining -= chunk

            log.info(f"Splitting ${amount_usd:,.0f} into {len(chunks)} chunks")
            for i, chunk in enumerate(chunks):
                result = await self._execute_single_swap(pair, side, chunk, slippage_bps, mev_risk)
                results.append(result)
                log.info(f"Chunk {i+1}/{len(chunks)} executed: {result.tx_hash[:10]}...")
                if i < len(chunks) - 1:
                    await asyncio.sleep(SPLIT_DELAY_S)
        else:
            result = await self._execute_single_swap(pair, side, amount_usd, slippage_bps, mev_risk)
            results.append(result)

        total_output = sum(r.output_amount for r in results)
        avg_slippage = sum(r.slippage_bps for r in results) / len(results)
        log.info(f"Swap complete. Total output: {total_output:.4f}, avg slippage: {avg_slippage:.0f} bps")
        return results


# Usage example
async def main():
    async with ProtectedSwapAgent(API_KEY) as agent:
        # Buy $25,000 of ETH with full MEV protection
        results = await agent.swap(
            pair="ETH/USDC",
            side="buy",
            amount_usd=25_000,
        )
        for r in results:
            print(f"tx: {r.tx_hash}")
            print(f"  output: {r.output_amount:.4f} ETH")
            print(f"  slippage: {r.slippage_bps} bps")
            print(f"  MEV risk was: {r.mev_risk_at_submission}")


if __name__ == "__main__":
    asyncio.run(main())

import aiohttp
from dataclasses import dataclass
from typing import List, Optional

@dataclass
class SandwichEvidence:
    victim_tx: str
    frontrun_tx: str
    backrun_tx: str
    estimated_extraction_usd: float
    attacker_address: str


async def detect_sandwich_in_block(
    block_number: int,
    agent_address: str,
    web3_session: aiohttp.ClientSession,
    rpc_url: str,
) -> List[SandwichEvidence]:
    """
    Analyze a block for sandwich attacks against agent_address.

    Looks for pattern:
    [some_address buys TOKEN_A] → [agent_address swaps] → [some_address sells TOKEN_A]
    where some_address is the same in both surrounding txs.
    """
    async def rpc(method, params):
        async with web3_session.post(rpc_url, json={
            "jsonrpc": "2.0", "id": 1, "method": method, "params": params
        }) as r:
            data = await r.json()
            return data["result"]

    # Get full block with transactions
    block = await rpc("eth_getBlockByNumber", [hex(block_number), True])
    txs = block["transactions"]

    # Find agent's transactions in this block
    agent_txs = [
        (i, tx) for i, tx in enumerate(txs)
        if tx["from"].lower() == agent_address.lower()
    ]

    evidence = []
    for agent_idx, agent_tx in agent_txs:
        # Check transactions immediately before and after
        if agent_idx == 0 or agent_idx == len(txs) - 1:
            continue

        before_tx = txs[agent_idx - 1]
        after_tx  = txs[agent_idx + 1]

        # Sandwich pattern: same sender for before and after tx
        if before_tx["from"].lower() == after_tx["from"].lower():
            # Check that both are DEX interactions (simplified: non-zero input data)
            if len(before_tx["input"]) > 10 and len(after_tx["input"]) > 10:
                # Estimate extraction (requires tx receipt analysis; simplified here)
                log.warning(
                    f"SANDWICH DETECTED in block {block_number}:"
                    f" {before_tx['hash'][:10]}... → agent → {after_tx['hash'][:10]}..."
                )
                evidence.append(SandwichEvidence(
                    victim_tx=agent_tx["hash"],
                    frontrun_tx=before_tx["hash"],
                    backrun_tx=after_tx["hash"],
                    estimated_extraction_usd=-1,  # requires deeper receipt analysis
                    attacker_address=before_tx["from"],
                ))

    return evidence


async def mev_monitoring_loop(agent_address: str, rpc_url: str):
    """Continuously monitor recent blocks for MEV extraction events."""
    last_block = 0
    total_events = 0

    async with aiohttp.ClientSession() as session:
        while True:
            try:
                # Get current block
                async with session.post(rpc_url, json={
                    "jsonrpc": "2.0", "id": 1,
                    "method": "eth_blockNumber", "params": []
                }) as r:
                    data = await r.json()
                    current_block = int(data["result"], 16)

                if current_block > last_block:
                    sandwiches = await detect_sandwich_in_block(
                        current_block, agent_address, session, rpc_url
                    )
                    total_events += len(sandwiches)
                    if total_events > 5:
                        log.error(
                            f"Agent sandwiched {total_events} times recently."
                            f" CONSIDER enabling Purple Flea MEV protection or"
                            f" switching to private RPC submission."
                        )
                    last_block = current_block

            except Exception as e:
                log.error(f"MEV monitor error: {e}")

            await asyncio.sleep(13)  # ~1 ETH block time