┌─────────────────────────────────────────────────────────────────────────────┐
│                         SANDWICH BOT DATA FLOW                              │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ┌──────────┐    ┌──────────────┐    ┌─────────────┐    ┌──────────────┐  │
│  │  RPC     │───▶│  MEMPOOL     │───▶│  OPPORTUNITY│───▶│  BUNDLE      │  │
│  │  (Erigon)│    │  MONITOR     │    │  CALCULATOR │    │  BUILDER     │  │
│  └──────────┘    └──────────────┘    └─────────────┘    └──────────────┘  │
│       │                │                    │                   │         │
│       ▼                ▼                    ▼                   ▼         │
│  newPendingTx    filter: large swaps    simulate:           construct:    │
│  subscription    on Uniswap V3           - optimal in      - frontrun tx │
│  (ws)            (amount > $50k)           - slippage        - victim tx  │
│                                       - gas est.        - backrun tx   │
│                                                                             │
│                                                          ┌──────────────┐  │
│                                                          │  FLASHBOTS   │  │
│                                                          │  RELAY       │  │
│                                                          │  (mev-share) │  │
│                                                          └──────────────┘  │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

// src/mempool/monitor.rs
use ethers::{
    providers::{Provider, Ws, Middleware, PubsubClient},
    types::{Transaction, H256, U256},
    prelude::*,
};
use futures::StreamExt;
use tokio::sync::mpsc;
use tracing::{info, warn, debug};
use std::sync::Arc;

const UNISWAP_V3_ROUTER: &str = "0xE592427A0AEce92De3Edee1F18E0157C05861564";
const MIN_SWAP_USD: U256 = U256::from(50_000_000_000_000_000_000u128); // $50k in wei (18 decimals)

pub struct MempoolMonitor {
    provider: Arc<Provider<Ws>>,
    tx_sender: mpsc::Sender<CandidateTx>,
    known_pools: Arc<DashMap<H160, PoolState>>,
}

#[derive(Debug, Clone)]
pub struct CandidateTx {
    pub hash: H256,
    pub from: H160,
    pub to: H160,
    pub input: Bytes,
    pub value: U256,
    pub gas_price: U256,
    pub gas_limit: U256,
    pub pool_address: H160,
    pub token_in: H160,
    pub token_out: H160,
    pub amount_in: U256,
    pub decoded: DecodedSwap,
}

impl MempoolMonitor {
    pub async fn new(
        ws_url: &str,
        tx_sender: mpsc::Sender<CandidateTx>,
        known_pools: Arc<DashMap<H160, PoolState>>,
    ) -> eyre::Result<Self> {
        let provider = Arc::new(Provider::<Ws>::connect(ws_url).await?);
        Ok(Self { provider, tx_sender, known_pools })
    }

    pub async fn run(&self) -> eyre::Result<()> {
        let mut stream = self.provider.subscribe_pending_txs().await?;

        while let Some(tx_hash) = stream.next().await {
            let provider = self.provider.clone();
            let sender = self.tx_sender.clone();
            let pools = self.known_pools.clone();

            tokio::spawn(async move {
                if let Err(e) = process_pending_tx(provider, sender, pools, tx_hash).await {
                    debug!("Failed to process tx {:?}: {}", tx_hash, e);
                }
            });
        }
        Ok(())
    }
}

async fn process_pending_tx(
    provider: Arc<Provider<Ws>>,
    sender: mpsc::Sender<CandidateTx>,
    pools: Arc<DashMap<H160, PoolState>>,
    tx_hash: H256,
) -> eyre::Result<()> {
    // Get full transaction - this is the latency critical path
    let tx = match provider.get_transaction(tx_hash).await? {
        Some(tx) => tx,
        None => return Ok(()), // Already mined or dropped
    };

    // Fast path: filter router interactions only
    if tx.to != Some(UNISWAP_V3_ROUTER.parse()?) {
        return Ok(());
    }

    // Decode calldata - only exactInputSingle and exactOutputSingle
    let decoded = match decode_swap_calldata(&tx.input)? {
        Some(d) => d,
        None => return Ok(()),
    };

    // Check pool exists in our registry
    let pool_addr = get_pool_address(decoded.token_in, decoded.token_out, decoded.fee)?;
    if !pools.contains_key(&pool_addr) {
        return Ok(());
    }

    // Estimate USD value - quick price oracle check
    let amount_usd = estimate_usd_value(decoded.token_in, decoded.amount_in).await?;
    if amount_usd < MIN_SWAP_USD {
        return Ok(());
    }

    let candidate = CandidateTx {
        hash: tx_hash,
        from: tx.from,
        to: tx.to.unwrap(),
        input: tx.input,
        value: tx.value,
        gas_price: tx.gas_price.unwrap_or_default(),
        gas_limit: tx.gas,
        pool_address: pool_addr,
        token_in: decoded.token_in,
        token_out: decoded.token_out,
        amount_in: decoded.amount_in,
        decoded,
    };

    // Send to opportunity calculator - non-blocking
    let _ = sender.try_send(candidate);
    Ok(())
}

fn decode_swap_calldata(input: &Bytes) -> eyre::Result<Option<DecodedSwap>> {
    // 0x414bf389 = exactInputSingle((address,address,uint24,address,uint256,uint256,uint160))
    // 0xdb3e2198 = exactOutputSingle((address,address,uint24,address,uint256,uint256,uint160))
    if input.len() < 4 { return Ok(None); }

    let selector = &input[0..4];
    match selector {
        [0x41, 0x4b, 0xf3, 0x89] => decode_exact_input_single(&input[4..]),
        [0xdb, 0x3e, 0x21, 0x98] => decode_exact_output_single(&input[4..]),
        _ => Ok(None),
    }
}

rust
// src/strategy/calculator.rs
use ethers::types::{U256, I256, H160};
use uniswap_v3_math::{sqrt_price_math, tick_math, swap_math};
use std::cmp::Ordering;

#[derive(Debug, Clone, Copy)]
pub struct PoolState {
    pub sqrt_price_x96: U256,
    pub tick: i32,
    pub liquidity: U128,
    pub fee: u32, // 500 = 0.05%, 3000 = 0.3%, 10000 = 1%
    pub token0: H160,
    pub token1: H160,
}

#[derive(Debug, Clone)]
pub struct SandwichParams {
    pub frontrun_amount_in: U256,
    pub backrun_amount_out: U256,
    pub expected_profit_usd: f64,
    pub gas_estimate: U256,
    pub victim_tx: CandidateTx,
}

pub struct OpportunityCalculator {
    pool_states: Arc<DashMap<H160, PoolState>>,
    token_prices: Arc<DashMap<H160, f64>>, // USD per token (18 decimals)
    config: CalculatorConfig,
}

#[derive(Debug, Clone)]
pub struct CalculatorConfig {
    pub min_profit_usd: f64,      // $2.50
    pub max_gas_price_gwei: u64,  // 15 gwei ceiling
    pub gas_limit: u64,           // 210,000
    pub max_slippage_bps: u32,    // 50 bps = 0.5%
}

impl OpportunityCalculator {
    pub fn new(
        pool_states: Arc<DashMap<H160, PoolState>>,
        token_prices: Arc<DashMap<H160, f64>>,
        config: CalculatorConfig,
    ) -> Self {
        Self { pool_states, token_prices, config }
    }

    pub fn calculate(&self, victim: &CandidateTx) -> Option<SandwichParams> {
        let pool = self.pool_states.get(&victim.pool_address)?;
        let pool = pool.value();

        // Determine swap direction: token_in -> token_out
        let zero_for_one = victim.token_in == pool.token0;

        // Victim's swap parameters
        let victim_amount_in = victim.amount_in;
        let victim_sqrt_price_limit = if zero_for_one {
            U256::from(0) // No limit
        } else {
            U256::MAX
        };

        // Simulate victim swap to get post-victim state
        let (victim_amount_out, post_victim_sqrt_price, post_victim_liquidity, _) = 
            swap_math::compute_swap_step(
                pool.sqrt_price_x96,
                pool.tick,
                pool.liquidity,
                victim_amount_in,
                pool.fee,
                zero_for_one,
            ).ok()?;

        // Now we need to find optimal frontrun amount
        // This is a convex optimization problem - ternary search on amount_in
        let optimal_frontrun = self.find_optimal_frontrun(
            pool,
            zero_for_one,
            victim_amount_in,
            post_victim_sqrt_price,
            post_victim_liquidity,
        )?;

        // Calculate backrun: we sell what we bought
        let (backrun_amount_out, _, _, _) = swap_math::compute_swap_step(
            post_victim_sqrt_price,
            // tick after victim - need to recalculate
            tick_math::get_tick_at_sqrt_ratio(post_victim_sqrt_price).ok()?,
            post_victim_liquidity,
            optimal_frontrun, // We're selling the token we bought
            pool.fee,
            !zero_for_one, // Opposite direction
        ).ok()?;

        // Profit calculation
        let token_out_price = self.token_prices.get(&victim.token_out)?.value();
        let profit_usd = (backrun_amount_out.as_u128() as f64 / 1e18) * token_out_price;

        // Gas cost
        let gas_cost_usd = self.estimate_gas_cost_usd();

        let net_profit = profit_usd - gas_cost_usd;

        if net_profit < self.config.min_profit_usd {
            return None;
        }

        // Check gas price ceiling
        let current_gas_gwei = victim.gas_price.as_u64() / 1_000_000_000;
        if current_gas_gwei > self.config.max_gas_price_gwei {
            return None;
        }

        Some(SandwichParams {
            frontrun_amount_in: optimal_frontrun,
            backrun_amount_out,
            expected_profit_usd: net_profit,
            gas_estimate: U256::from(self.config.gas_limit),
            victim_tx: victim.clone(),
        })
    }

    fn find_optimal_frontrun(
        &self,
        pool: &PoolState,
        zero_for_one: bool,
        victim_amount_in: U256,
        post_victim_sqrt_price: U256,
        post_victim_liquidity: U128,
    ) -> Option<U256> {
        // Ternary search on frontrun amount
        // Search space: 1 wei to victim_amount_in * 2 (capped by pool liquidity)
        let max_frontrun = std::cmp::min(
            victim_amount_in * 2,
            pool.liquidity.into(),
        );

        let mut left = U256::from(1);
        let mut right = max_frontrun;

        // 50 iterations = precision of ~1 wei for 2^160 range
        for _ in 0..50 {
            if right <= left { break; }

            let third = (right - left) / 3;
            let m1 = left + third;
            let m2 = right - third;

            let profit1 = self.simulate_sandwich_profit

#MEV #Rust #Arbitrum #Flashbots