Setting Up a Real-Time MEV Detection and Alert System

Maximal Extractable Value (MEV) represents profit that can be extracted by reordering, including, or censoring transactions within blocks. A real-time detection system monitors the public mempool for patterns like arbitrage, liquidations, and sandwich attacks. Setting up such a system requires connecting to a node, streaming pending transactions, and analyzing them for specific MEV signals. This tutorial will guide you through building a basic detection and alert pipeline using Ethereum and common Web3 tools.

The core of the system is a connection to a node's JSON-RPC endpoint. You need access to the eth_getBlockByNumber and a WebSocket connection for newPendingTransactions. Services like Alchemy, Infura, or running your own Geth or Erigon node provide this. The following Python snippet establishes a WebSocket connection using the web3.py library:

pythonCopy
from web3 import Web3
# Connect to WebSocket endpoint (Alchemy/Infura)
w3 = Web3(Web3.WebsocketProvider('wss://eth-mainnet.g.alchemy.com/v2/YOUR_KEY'))
# Subscribe to pending transactions
def handle_event(event):
    tx_hash = event['params']['result']
    tx = w3.eth.get_transaction(tx_hash)
    analyze_transaction(tx)

Detection logic analyzes transaction data. For a sandwich attack, look for a user's swap transaction with a low maxPriorityFee surrounded by two attacker transactions with higher fees. For arbitrage, monitor for profitable DEX price differences across pools like Uniswap and Sushiswap. Your analyze_transaction function should decode the input data to identify the target contract (e.g., Uniswap V3 Router) and simulate the transaction's effect using a tool like Tenderly or Ethylene. Key data points include token addresses, amounts, and expected profit.

Once a potential MEV opportunity or threat is identified, you need an alert system. This can be as simple as logging to a file, sending a message to a Discord webhook, or triggering an automated trading bot. For security monitoring, alerts for suspicious frontrunning of your own transactions are critical. A simple Discord alert using requests might look like:

pythonCopy
import requests
discord_webhook_url = 'YOUR_WEBHOOK_URL'
alert_message = {
    'content': f'⚠️ Potential Sandwich Attack Detected on TX: {tx_hash}'
}
requests.post(discord_webhook_url, json=alert_message)

For a production system, consider scalability and data persistence. Processing every pending transaction can be overwhelming. Implement filtering by gas price, specific contract addresses, or transaction value to reduce load. Use a database like PostgreSQL or TimescaleDB to store detected events for later analysis. Frameworks like Apache Kafka or RabbitMQ can help manage the transaction stream. Always be mindful of rate limits on your node provider and the ethical implications of the MEV you choose to pursue or mitigate.

This setup provides a foundation. Advanced systems incorporate Flashbots MEV-Share data to see private order flow, use EigenPhi-style analytics for complex MEV classification, or even run simulations with forked mainnet state. The goal is to create a transparent monitoring tool that can protect users from malicious MEV or help researchers understand this complex ecosystem. Remember, the mempool is a competitive space, and the most profitable opportunities require ultra-low latency and sophisticated strategy.

A robust MEV detection pipeline requires a reliable connection to the blockchain. The core component is an Ethereum execution client like Geth or Erigon. You will need to run a full node or, more practically, connect to a high-performance node provider via WebSocket. Services like Alchemy, Infura (with Websockets enabled), or QuickNode offer the low-latency, high-throughput JSON-RPC connections necessary for streaming pending transactions from the mempool. A standard HTTPS endpoint is insufficient for real-time alerts; you must use the wss:// WebSocket protocol.

Your development environment must support asynchronous programming to handle continuous data streams. We recommend using Node.js (v18+) with libraries like ethers.js v6 or web3.js v4 for interacting with the blockchain. For Python-based systems, web3.py with asyncio is a common choice. You will also need a package manager (npm or pip) and a code editor like VS Code. Basic familiarity with terminal commands and environment variable management (using a .env file) is assumed.

To detect MEV opportunities, you need more than raw transaction data. You must integrate with specialized MEV data services that classify and label transactions. The Flashbots MEV-Share API provides insights into bundled transactions and searcher activity. For a broader view, EigenPhi and Etherscan's API (for contract verification) are valuable. Your system will need API keys for these services, and you should implement proper rate limiting and error handling for each external dependency.

Real-time detection demands a backend capable of processing and analyzing transactions as they arrive. A simple script can work for prototyping, but for production, consider a message queue or stream processor like Redis Pub/Sub, Apache Kafka, or RabbitMQ to decouple data ingestion from analysis logic. This allows you to scale your alerting engine independently. You will also need a database to log detected events; PostgreSQL or TimescaleDB are suitable for time-series data on arbitrage opportunities or liquidations.

Finally, define your alerting mechanism. This could be a simple console log, but for actionable alerts, integrate with communication platforms. Common choices include sending messages via a Discord webhook, Telegram Bot API, or Slack API. For more advanced systems, you might trigger automated responses through smart contracts or off-chain agents. Ensure you have the necessary permissions and webhook URLs configured for your chosen notification channel before deploying your detector.

A real-time MEV detection system operates by streaming pending transactions from a node's mempool, analyzing them for profitable patterns, and triggering alerts or automated actions. The foundational architecture typically consists of three primary layers: the Data Ingestion Layer, which connects to blockchain nodes; the Processing & Analysis Layer, where transaction bundles are simulated and opportunities are identified; and the Alerting & Action Layer, which delivers findings to users or execution systems. Each layer must be designed for low latency to compete in the sub-second MEV landscape.

The Data Ingestion Layer is responsible for acquiring raw transaction data. This involves establishing a WebSocket connection to an Ethereum execution client like Geth or Erigon to subscribe to the newPendingTransactions event. For production reliability and redundancy, you should connect to multiple node providers (e.g., Alchemy, Infura, or a private node cluster) to avoid missing transactions due to provider-specific filtering or downtime. Ingested transactions must be parsed to decode basic fields like to, from, value, and input data, which is the first step in filtering the firehose of pending data.

At the heart of the system is the Processing & Analysis Layer. Here, raw transactions are grouped into potential arbitrage bundles or liquidations based on heuristics. For example, detecting a large DEX swap on Uniswap V3 might trigger a simulation of a back-running arbitrage opportunity on Sushiswap. This requires a local simulation environment using tools like eth_call to a forked network state or the Tenderly Simulation API. The simulation estimates gas costs and potential profit, filtering out unprofitable opportunities. Complex strategies may require analyzing transaction ordering within a block.

Finally, the Alerting & Action Layer delivers results. For a detection-only system, this could involve sending alerts via Discord webhooks, Telegram bots, or SMS. An automated searcher system would take this further by constructing a valid transaction bundle and submitting it to a relay like Flashbots Protect or the bloXroute Gateway. The alert payload must include critical details: the target contract address, the expected profit in ETH/USD, the required gas, and a unique opportunity ID. All system components should be logged meticulously for post-mortem analysis of missed opportunities or failed executions.

When architecting this system, key technical challenges include managing data volume (the mempool can process 10,000+ transactions per minute), ensuring simulation accuracy against a rapidly changing state, and minimizing end-to-end latency. A well-designed pipeline might process and analyze a transaction within 200-400 milliseconds of its broadcast. Using a message queue like Apache Kafka or Redis Pub/Sub between layers can help decouple components and handle load spikes effectively.

To begin development, you can use libraries like ethers.js or web3.py for node interaction, the Ethers.js Flashbots Bundle provider for bundle submission, and Tenderly or a local Hardhat fork for simulations. Start by building a simple alert for large pending transfers, then incrementally add complexity for DEX arbitrage or liquidation detection. Remember that MEV is a competitive field; your system's speed and reliability will directly determine its profitability.

A mempool listener is a service that connects to one or more Ethereum nodes (e.g., via Geth or Erigon) and subscribes to the pending transaction stream. This provides a real-time feed of all transactions broadcast to the network before they are included in a block. For MEV detection, you need access to a full node or a specialized RPC provider that supports the eth_subscribe method for pending transactions. Services like Alchemy, Infura, and QuickNode offer enhanced mempool access, which is more reliable than running your own public node.

The technical implementation involves establishing a WebSocket connection to the node's RPC endpoint. You will listen for the newPendingTransactions event. Each transaction hash received must then be fetched in full using eth_getTransactionByHash to obtain its details: sender (from), target contract (to), calldata (input), value, and gas parameters. It's critical to handle the high volume and rate of transactions; a robust listener implements connection error handling, re-subscription logic, and a queueing system (e.g., using Redis or RabbitMQ) to decouple ingestion from processing.

For effective MEV detection, you must listen on multiple networks simultaneously. Opportunities like cross-chain arbitrage or liquidations can occur on Layer 2s like Arbitrum or Optimism just as they do on Ethereum Mainnet. Your listener should be architected to manage separate connections for each chain, tagging each transaction with its chainId. This setup allows your detection engine to analyze opportunities that involve asset movements between these interconnected environments.

Here is a basic Python example using the Web3.py library to start a mempool listener:

pythonCopy
from web3 import Web3
import asyncio

w3 = Web3(Web3.WebsocketProvider('wss://mainnet.infura.io/ws/v3/YOUR_API_KEY'))

def handle_transaction(tx_hash):
    try:
        tx = w3.eth.get_transaction(tx_hash)
        print(f"New TX: {tx['hash'].hex()} from {tx['from']} to {tx['to']}")
        # Add tx to processing queue
    except Exception as e:
        print(f"Error fetching tx {tx_hash}: {e}")

async def log_loop():
    pending_filter = w3.eth.filter('pending')
    while True:
        for tx_hash in pending_filter.get_new_entries():
            handle_transaction(tx_hash)
        await asyncio.sleep(0.1)

# Start the loop
asyncio.run(log_loop())

This simple script captures transaction hashes. A production system would replace the print statement with a call to a queue or database.

The raw transaction data is just the beginning. To identify MEV, you must decode the calldata to understand the intent. This requires integrating with ABI databases or Etherscan-like services to resolve function signatures. For unknown or new contracts, you may need to attempt bytecode analysis or trace simulation. The listener's role is to reliably capture and forward this raw data with minimal latency to the next stage: the transaction analysis and simulation engine.

A detection heuristic is a piece of logic that analyzes transaction data to identify a specific MEV pattern. It's the core of your monitoring system. Heuristics typically examine a mempool transaction or a confirmed block for known signatures, such as a sandwich attack, a large arbitrage opportunity, or a specific exploit vector. The goal is to write code that can programmatically flag these events in real-time.

Start by defining the pattern you want to detect. For a classic sandwich attack, your heuristic would scan for a victim transaction with a high slippage tolerance, then look for a prior attacker transaction that front-runs it with the same swap path, and a following transaction that back-runs it to profit from the induced price movement. You implement this by parsing transaction calldata, tracking Uniswap pool reserves, and calculating potential profit.

Write your heuristics in a structured, modular way. A common pattern is to create a base Heuristic class or interface, then extend it for specific strategies like SandwichHeuristic, ArbitrageHeuristic, or LiquidationHeuristic. Each heuristic should have a run(tx: Transaction) method that returns a Finding object if the condition is met. This Finding should contain all relevant data: the transaction hash, the attacker's address, the protocol involved, the estimated profit, and a severity level.

Use existing frameworks and libraries to simplify development. The Forta Network provides a robust TypeScript/JavaScript SDK for writing detection bots. EigenPhi and Flashbots offer open-source research and data on MEV transaction patterns which can inform your logic. Always test your heuristics against historical blockchain data using a provider like Alchemy or QuickNode to fine-tune parameters and reduce false positives.

Finally, consider the performance and cost of your heuristics. Scanning every transaction in the mempool is resource-intensive. Optimize by filtering for transactions interacting with specific high-value protocols (e.g., Uniswap V3, Aave, Compound) or above a certain gas price threshold. Your heuristic should execute quickly to avoid lag in the alerting pipeline, which is critical for time-sensitive MEV opportunities.

The block analysis configuration defines the specific MEV patterns your system will monitor. Chainscore's /v1/mev/block-analysis endpoint processes raw block data to identify suspicious transactions. You must configure two primary detection categories: sandwich attacks and arbitrage opportunities. A sandwich attack occurs when a malicious transaction is placed before and after a victim's DEX swap to profit from price impact. Arbitrage involves exploiting price differences for the same asset across different liquidity pools or exchanges within the same block.

To begin, you will call the API with a JSON payload specifying the block range and detection parameters. The key is the detectionConfig object. For sandwich detection, you typically enable sandwichDetection and can set a minimum profit threshold in USD to filter out negligible attacks. For arbitrage, enable arbitrageDetection and specify which DEXes (like Uniswap V3, Curve, Balancer) to monitor. You can also configure includeFailedTxs to analyze transactions that reverted, as failed MEV attempts are common and informative.

Here is a basic configuration example for analyzing a single block on Ethereum mainnet:

jsonCopy
{
  "chainId": 1,
  "blockNumber": 19500000,
  "detectionConfig": {
    "sandwichDetection": {
      "enabled": true,
      "minProfitUsd": 50
    },
    "arbitrageDetection": {
      "enabled": true,
      "dexes": ["uniswapv3", "sushiswap"]
    },
    "includeFailedTxs": true
  }
}

This request instructs the system to scan block 19500000 for sandwich attacks netting over $50 and arbitrage loops on Uniswap V3 and Sushiswap, including reverted transactions in the analysis.

The API response will contain a detectedActivities array. Each activity object includes the type (e.g., SANDWICH), the involved transactions (frontrun, victim, backrun), calculated profit in USD and the token addresses, and the dexPool where it occurred. For a sandwich attack, the victimTransactionHash is a critical field for your alert system. You should log this data and trigger an alert—such as a Discord webhook or PagerDuty event—whenever an activity is detected, especially if the profit exceeds a critical threshold you define.

For continuous monitoring, you must integrate this block analysis call into a service that polls for new blocks. A common pattern is to use a WebSocket connection to a node provider for new block headers, then immediately submit the block number to Chainscore's API for analysis. Ensure your service handles rate limits and queues blocks during high-throughput periods. The analysis latency is typically under 2 seconds, enabling near real-time alerts. You can find the complete specification and additional parameters in the official Block Analysis API documentation.

Effective configuration requires tuning. Start with broad detection (all DEXes, low profit threshold) and analyze the output for a day. You will notice patterns and noise. Gradually increase minProfitUsd to focus on economically significant attacks and curate the dexes list to the pools most relevant to your monitored protocols. This iterative tuning reduces alert fatigue and ensures your system highlights the MEV activity that poses genuine risk or presents strategic opportunities for your project.

An alert system is only as good as its delivery mechanism. After your detection logic identifies a potential MEV event, you need to notify the right people instantly. This step involves integrating with external notification services to push alerts to platforms your team monitors, such as Slack, Discord, Telegram, or email. The goal is to minimize the time between detection and human review, enabling rapid investigation and potential intervention. For critical systems, consider implementing multiple, redundant channels to ensure no alert is missed due to a single point of failure.

Most alerting workflows follow a simple pattern: your detection script or bot formats a message with key details—like the transaction hash, involved addresses, estimated profit, and attack type—and sends it via an API. For Slack, you would use Incoming Webhooks to post to a specific channel. Discord offers similar functionality through Webhooks. For Telegram, you interact with the Bot API. A basic Python example using the requests library for a Slack webhook might look like this:

pythonCopy
import requests
import json

WEBHOOK_URL = "https://hooks.slack.com/services/..."

def send_slack_alert(tx_hash, alert_type, profit_eth):
    message = {
        "text": f"🚨 MEV Alert: {alert_type}",
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Transaction*: `{tx_hash}`\n*Type*: {alert_type}\n*Estimated Profit*: {profit_eth} ETH"
                }
            }
        ]
    }
    requests.post(WEBHOOK_URL, json=message)

When designing your alerts, prioritize actionable information. Each notification should allow an analyst to quickly understand the severity and context. Include direct links to block explorers (e.g., Etherscan), MEV explorer dashboards like EigenPhi or Ethereum.org's Block Explorer, and any relevant internal tools. For sandwich attacks, link to the victim's transaction and the attacker's profitable arbitrage. For liquidation bots, show the vault address and the health factor. Structuring data this way turns an alert from a simple notification into a direct launchpad for investigation.

It's crucial to implement alert deduplication and rate limiting to prevent notification fatigue. If your detector scans every new block, a single, ongoing MEV attack could generate dozens of identical alerts. Implement logic to group related events (e.g., all transactions from the same searcher address within a 10-block window) or suppress follow-up alerts for a cooldown period after the first notification. This ensures your team is alerted to new incidents without being overwhelmed by noise, maintaining the system's credibility and urgency.

Your system is now monitoring for common MEV strategies like sandwich attacks, liquidations, and arbitrage. The next step is to refine your detection logic. Start by analyzing the false positive rate from your initial runs. Adjust your heuristics—like the minimum profit threshold in your analyze_transaction function or the gas price multiplier for frontrunning detection—based on real network data. Consider implementing a machine learning model to classify transactions if rule-based logic becomes insufficient. Open-source datasets like the Flashbots MEV-Share data can be valuable for training.

To scale and harden your system, focus on infrastructure. Replace the local SQLite database with a time-series database like TimescaleDB or QuestDB for efficient storage and querying of high-volume mempool data. Implement a message queue (e.g., RabbitMQ or Apache Kafka) to decouple the listener from the analyzer, ensuring you don't drop transactions during processing spikes. For production, run your services in containers using Docker and orchestrate them with Kubernetes or a similar tool for resilience and easy scaling across multiple blockchain nodes.

Finally, expand your monitoring and response capabilities. Integrate with on-chain automation platforms like Gelato Network or Chainlink Automation to create conditional execution bots that can counter detrimental MEV or capitalize on opportunities you identify. Continuously monitor the Ethereum Execution API specification and client updates (Geth, Erigon, Nethermind) for changes to transaction propagation and mempool access. The MEV landscape evolves rapidly; maintaining an effective system requires ongoing adaptation to new strategies like PBS (Proposer-Builder Separation) and the techniques emerging in its wake.