Sandwich Detector

📁 File Structure 10 files

sandwich-detector/ ├── LICENSE ├── README.md ├── pyproject.toml ├── security-notes.md ├── src/ │ └── sandwich_detector/ │ ├── __init__.py │ ├── alerts.py │ ├── detector.py │ ├── mempool.py │ └── types.py └── tests/ └── test_detector.py

📖 Documentation Preview README excerpt

Real-time mempool monitor detecting sandwich attacks on DEX swaps.

Price: $7.99 | License: MIT | Python: >=3.10

Overview

A real-time sandwich attack detection engine that monitors pending transactions in the Ethereum mempool. Identifies front-run/victim/back-run patterns targeting DEX swaps using gas price analysis, timing heuristics, and token pair matching.

Architecture


Mempool (WebSocket)
      │
      ▼
MempoolMonitor ──→ Decode swap calldata
      │
      ▼
SandwichDetector ──→ Pattern matching
      │                 │
      │            ┌────┴─────┐
      │            │ Heuristics│
      │            │ - Gas     │
      │            │ - Timing  │
      │            │ - Pairs   │
      │            └────┬─────┘
      ▼                 ▼
AlertManager ──→ Console / File / Webhook

Features

WebSocket-based mempool subscription (geth, Erigon, Nethermind)
Heuristic sandwich detection: gas price, timing, token pair matching
Configurable alert levels (LOW / MEDIUM / HIGH / CRITICAL)
Confidence scoring (0.0 - 1.0)
Multiple alert channels: console, JSONL file, Slack/Discord webhook
Known DEX router database (Uniswap V2/V3, SushiSwap, 1inch)
Decorator-based event handlers
Detection statistics tracking

Quick Start


pip install -e .


import asyncio
from sandwich_detector import SandwichDetector, MempoolMonitor, DetectorConfig

config = DetectorConfig(rpc_url="ws://localhost:8546")
monitor = MempoolMonitor(config)
detector = SandwichDetector(config)

@monitor.on_swap
def handle_swap(swap):

*... continues with setup instructions, usage examples, and more.*

📄 Code Sample .py preview

src/sandwich_detector/alerts.py """Alerting module for sandwich attack notifications.""" import json import time import logging from typing import Optional from .types import SandwichAttack, AlertLevel logger = logging.getLogger(__name__) class AlertManager: """ Manages alerts for detected sandwich attacks. Supports console logging, file logging, and webhook notifications. Example:: from sandwich_detector.alerts import AlertManager alert_mgr = AlertManager( webhook_url="https://hooks.slack.com/...", log_file="sandwiches.jsonl", ) alert_mgr.send(attack) """ def __init__( self, webhook_url: Optional[str] = None, log_file: Optional[str] = None, min_alert_level: AlertLevel = AlertLevel.MEDIUM, ) -> None: self._webhook_url = webhook_url self._log_file = log_file self._min_alert_level = min_alert_level self._alert_count = 0 self._level_order = [AlertLevel.LOW, AlertLevel.MEDIUM, AlertLevel.HIGH, AlertLevel.CRITICAL] def send(self, attack: SandwichAttack) -> None: """Send an alert for a detected sandwich attack.""" if not self._should_alert(attack.alert_level): return self._alert_count += 1 # Console log self._log_console(attack) # ... 80 more lines ...