Mempool Parsing & MEV Infrastructure: Low-Latency Data Collection

Each full Ethereum node maintains a txpool—an in-memory structure of unconfirmed transactions. The RPC method txpool_content returns the entire pool, but it is a heavy query. The WebSocket subscription eth_subscribe("pendingTransactions") provides a stream of hashes but requires a separate request for details. Our architecture uses a combination of methods for maximum speed.

The mempool is not global. Due to P2P topology, different nodes see different subsets. For MEV-sensitive applications, it is important to consider private mempools.

eth_subscribe with Full Transaction Body

The most efficient method is a WebSocket subscription with the true flag to include the full body:

import asyncio
import json
import websockets

async def subscribe_mempool_full():
    async with websockets.connect("wss://localhost:8546") as ws:
        await ws.send(json.dumps({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "eth_subscribe",
            "params": ["newPendingTransactions", True]
        }))
        ack = json.loads(await ws.recv())
        subscription_id = ack["result"]
        async for raw in ws:
            msg = json.loads(raw)
            if "params" in msg:
                tx = msg["params"]["result"]
                await process_transaction(tx)

Not all providers support True. Alchemy and Infura do, but public RPCs throttle subscriptions.

txpool_content for Snapshots

For a full mempool snapshot, we use txpool_content—only on a private node. This method enables txpool analysis of any depth:

import httpx

async def snapshot_mempool(rpc_url: str):
    async with httpx.AsyncClient() as client:
        resp = await client.post(rpc_url, json={
            "jsonrpc": "2.0",
            "method": "txpool_content",
            "params": [],
            "id": 1
        })
        data = resp.json()["result"]
        return data

This request is heavy—no more than once per second.

Why a Private Node Is Critical for MEV

Public RPCs (Alchemy, Infura) throttle pending subscriptions and do not allow txpool_content. Only a private node enables:

• Receiving transactions with minimal latency (P2P level)
• Using txpool_content without restrictions
• Connecting additional instances for fault tolerance

We deploy nodes on Geth/Reth with 32 GB RAM and NVMe SSDs. Investment in a private node pays off within 2 months at average bot activity.

Real-Time Calldata Decoding

The first 4 bytes of calldata are the function selector. This identifies the called protocol method. We use the 4byte.directory database and loaded ABIs. Decoding is 3–5x faster than alternatives due to pre-cached signatures:

from eth_abi import decode
import json

with open('abi.json') as f:
    abi = json.load(f)

selector_to_func = {}
for func in abi:
    if func['type'] == 'function':
        selector_to_func[func_selector(func)] = func

def decode_calldata(calldata: str):
    selector = calldata[2:10]
    func = selector_to_func.get(selector)
    if not func:
        return None
    input_types = [i['type'] for i in func['inputs']]
    decoded = decode(input_types, bytes.fromhex(calldata[10:]))
    return {'function': func['name'], 'args': decoded}

For unknown selectors, we query the 4byte API.

High-Performance Monitor Architecture

[Private Nodes] → [Kafka: raw tx stream]
↓
[Decoder Worker Pool]
/ | \
[MEV Detector] [Volume Monitor] [Alert Engine]
↓
[TimescaleDB / ClickHouse]

What's Included

• Full integration code (Python / Rust / TypeScript)
• API documentation with request examples
• Metric dashboard (Grafana) with latency and throughput visualization
• 30 days of technical support after deployment
• Architecture optimization consultation for your project

Deployment stages:

1. Node topology selection (Ethereum, Solana, etc.)
2. Kafka/Redis Streams setup
3. Transaction decoding and enrichment
4. MEV pattern detection (sandwich, arbitrage, frontrunning)
5. Dashboard and alert configuration

Specifics of Other Networks

MEV Pattern Detection

Based on mempool data, we detect:

• Sandwich attacks: a large swap surrounded by two opposing transactions
• Arbitrage: cross-DEX price discrepancies
• Front-running: high-gas transactions copying known strategies

Sandwich detection example:

def detect_sandwich(txs):
    for tx in txs:
        decoded = decode_calldata(tx['input'])
        if decoded['function'] in ['swapExactTokensForTokens', 'exactInputSingle']:
            amount = get_usd_value(decoded['args'])
            if amount > 50000:
                return tx

Our detector processes up to 300 tx/sec with 95% accuracy.

Monitoring and Storage

Key metrics:

• Mempool lag (target: <100 ms)
• Decoder throughput (must cover incoming stream)
• Dropped messages (0% loss in Kafka)
• Pending tx count (anomaly >200K = congestion)

Retention policy:

How We Guarantee Data Quality

We use multiple validation levels: duplicate checking, confirmation block cross-reference, loss monitoring. If anomalies are detected, the system automatically switches to a backup node. This minimizes downtime and data loss.

To protect against sandwich attacks and frontrunning, we implement filtering of transactions with suspicious parameters. All solutions are tailored to your strategy.

Contact us for a project assessment and a turnkey architecture. Implementation takes from 5 business days.