Mempool Monitoring Bot Development

Mempool Monitoring Bot Development

Mempool is not just a transaction queue. It's a data source for intentions: what the market is about to do in the next block. MEV bots that earn on this data, there are hundreds running on Ethereum. But mempool monitoring is needed not just for arbitrage — liquidation bots for DeFi protocols, early attack detection systems, front-running protection, gas price analytics — all this is built on reading mempool in real-time.

Mempool Access: Three Levels

Level 1: Public mempool via RPC

The simplest way — eth_subscribe("newPendingTransactions") via WebSocket:

import { ethers } from 'ethers';

const wsProvider = new ethers.WebSocketProvider(
  `wss://eth-mainnet.g.alchemy.com/v2/${process.env.ALCHEMY_KEY}`
);

wsProvider.on('pending', async (txHash: string) => {
  const tx = await wsProvider.getTransaction(txHash);
  if (!tx) return;  // transaction could be dropped

  await processPendingTx(tx);
});

Problem: public RPC sees only part of mempool. A node reveals a pending transaction only after it distributes it to the network. Private transactions (via Flashbots, MEV Blocker, Cowswap) — not visible at all.

Level 2: Own node with txpool API

Run your own Geth/Reth with open txpool namespace:

# Geth with txpool API enabled
geth --mainnet \
  --http \
  --http.api eth,net,txpool \
  --ws \
  --ws.api eth,net,txpool \
  --txpool.globalslots 8192 \
  --txpool.accountslots 64

Now txpool_content is available — full mempool snapshot:

const txpoolContent = await provider.send('txpool_content', []);
// { pending: { [address]: { [nonce]: Transaction } }, queued: { ... } }

const txpoolStatus = await provider.send('txpool_status', []);
// { pending: '0x1234', queued: '0x56' }

Your own node sees all transactions that the p2p network passes through it. This is significantly more complete than public RPC.

Level 3: Node colocation + peering

For serious MEV or latency-critical tasks: hosting the node in the same data center as major validators (Equinix NY5, Amsterdam AMS3). Configuring maximum peers and connecting to specialized mesh networks:

# Connection to Bloxroute BDN (Blockchain Distribution Network)
geth --mainnet \
  --bootnodes "enode://bloxroute-peer@..." \
  --maxpeers 100

This reduces latency from ~200ms to ~50ms to receiving a transaction. Critical for front-running bots, but overkill for most monitoring tasks.

Transaction Decoding

Raw transaction is a set of bytes. To understand what it does, decode calldata:

import { Interface, FunctionFragment } from 'ethers';

// Load ABI of known protocols
const UNISWAP_V2_ABI = [...]; // swapExactTokensForTokens, etc.
const UNISWAP_V3_ABI = [...]; // exactInputSingle, etc.

const uniV2Iface = new Interface(UNISWAP_V2_ABI);
const uniV3Iface = new Interface(UNISWAP_V3_ABI);

function decodeUniswapTx(tx: ethers.TransactionResponse): DecodedSwap | null {
  if (!tx.data || tx.data === '0x') return null;

  const selector = tx.data.slice(0, 10);  // first 4 bytes

  // Try to decode as Uniswap V3 exactInputSingle
  if (selector === '0x414bf389') {
    try {
      const decoded = uniV3Iface.decodeFunctionData('exactInputSingle', tx.data);
      return {
        protocol: 'UniswapV3',
        tokenIn: decoded.params.tokenIn,
        tokenOut: decoded.params.tokenOut,
        amountIn: decoded.params.amountIn,
        amountOutMinimum: decoded.params.amountOutMinimum,
        recipient: decoded.params.recipient,
      };
    } catch { return null; }
  }

  return null;
}

For unknown contracts — 4byte.directory API to find signature by selector:

async function resolveSelector(selector: string): Promise<string | null> {
  const res = await fetch(`https://www.4byte.directory/api/v1/signatures/?hex_signature=${selector}`);
  const data = await res.json();
  return data.results[0]?.text_signature ?? null;
}

Usage Patterns: What People Actually Do with Mempool

Liquidation bot — monitoring positions on Aave/Compound near liquidation threshold:

// Listen to prices in mempool — Chainlink oracle updater transactions
async function watchChainlinkUpdates(tx: ethers.TransactionResponse) {
  if (tx.to !== CHAINLINK_AGGREGATOR) return;

  // Decode new price from calldata
  const newPrice = decodeOracleUpdate(tx.data);

  // Simulate liquidation with new price
  const liquidatablePositions = await simulateLiquidations(newPrice);

  if (liquidatablePositions.length > 0) {
    // Send liquidation transaction with higher gas price
    await sendLiquidationTx(liquidatablePositions[0], {
      maxFeePerGas: tx.maxFeePerGas! * 2n,  // outpace oracle
    });
  }
}

Sandwich attack detection — to protect your protocol:

interface SandwichCandidate {
  victimTx: string;
  frontRunTx: string;
  estimatedProfit: bigint;
}

async function detectPotentialSandwich(
  pendingTxs: Map<string, ethers.TransactionResponse>
): Promise<SandwichCandidate[]> {
  const candidates: SandwichCandidate[] = [];

  for (const [hash, tx] of pendingTxs) {
    const swap = decodeUniswapTx(tx);
    if (!swap) continue;

    // Look for already-sent transactions on same pool with higher gas
    const frontRuns = [...pendingTxs.values()].filter(other =>
      other.hash !== hash &&
      decodeUniswapTx(other)?.tokenIn === swap.tokenIn &&
      other.maxFeePerGas! > tx.maxFeePerGas!
    );

    if (frontRuns.length > 0) {
      candidates.push({ victimTx: hash, frontRunTx: frontRuns[0].hash, estimatedProfit: 0n });
    }
  }

  return candidates;
}

Gas price forecasting — analyzing gas distribution in mempool to predict next base fee:

async function analyzeGasDistribution() {
  const txpoolContent = await provider.send('txpool_content', []);
  const gasPrices: bigint[] = [];

  for (const addrTxs of Object.values(txpoolContent.pending)) {
    for (const tx of Object.values(addrTxs as any)) {
      gasPrices.push(BigInt((tx as any).maxFeePerGas || (tx as any).gasPrice));
    }
  }

  gasPrices.sort((a, b) => (a < b ? -1 : 1));

  return {
    p25: gasPrices[Math.floor(gasPrices.length * 0.25)],
    p50: gasPrices[Math.floor(gasPrices.length * 0.50)],  // median
    p75: gasPrices[Math.floor(gasPrices.length * 0.75)],
    p95: gasPrices[Math.floor(gasPrices.length * 0.95)],
    count: gasPrices.length,
  };
}

Architecture of High-Load Bot

On mainnet Ethereum mempool contains 50–200k transactions. Processing each synchronously is impossible:

WebSocket listener
      │ raw tx hash
      ▼
  Queue (Redis Streams)
      │
  Workers (x8 processes)
      │ decoding + filtering
      ▼
  Signal Bus (Redis Pub/Sub)
      │ only relevant events
      ▼
  Strategy Handlers
      │ liquidation / arb / alert
      ▼
  Execution Engine (with rate limiting)

Workers decode transactions in parallel, only what passes filter goes to signal queue. Execution engine manages transaction sending to avoid flooding network on mass events.

Flashbots and Private Transactions

Some transactions never enter public mempool — sent directly to validators via Flashbots MEV-Boost or Flashbots Protect. To monitor this traffic, you need a separate approach — subscribe to Flashbots event stream or partnership with block builders.

For most tasks (liquidations, protocol monitoring), public mempool is sufficient. For competitive MEV arbitrage — without Flashbots integration, it's hard to work: bots with private transactions have structural advantage.

Latency: Measurement and Optimization

class LatencyTracker {
  private samples: number[] = [];

  recordTxSeen(txHash: string, seenAt: number) {
    // seenAt — timestamp from WebSocket (performance.now())
    this.txSeenMap.set(txHash, seenAt);
  }

  recordTxMined(txHash: string, blockTimestamp: number) {
    const seenAt = this.txSeenMap.get(txHash);
    if (!seenAt) return;
    this.samples.push(Date.now() - seenAt);
  }

  getP50LatencyMs(): number {
    const sorted = [...this.samples].sort((a, b) => a - b);
    return sorted[Math.floor(sorted.length / 2)];
  }
}

Typical numbers: public Alchemy WebSocket — 150–400ms from broadcast to receipt. Own node with 50+ peers — 50–150ms. Node colocation in Equinix — 10–50ms.