721 lines
27 KiB
Python
721 lines
27 KiB
Python
"""
|
|
MEV Shield Analysis — Proactive Transaction Protection
|
|
========================================================
|
|
Assesses any pending or planned transaction for MEV extraction vulnerability
|
|
BEFORE it's sent. Identifies sandwich attack risk, frontrunning exposure,
|
|
and backrunning vulnerability. Provides actionable protection strategies.
|
|
|
|
The tool analyzes:
|
|
- MEMPOOL: Pending transaction patterns in the block's proximity
|
|
- SLIPPAGE: How much slippage tolerance invites MEV extraction
|
|
- GAS_PRICE: Whether gas bidding makes the tx a target
|
|
- PAIR_LIQUIDITY: How thin liquidity enables sandwich attacks
|
|
- TOKEN_PAIR: Historical MEV activity on this specific pair
|
|
- TIMING: Peak vs off-peak MEV activity periods
|
|
|
|
TOOL : mev_protection
|
|
TIER : premium
|
|
PRICE : $0.08 (80000 atoms)
|
|
TRIAL : 1 free check
|
|
|
|
Data Sources (all free):
|
|
- Local MEV database (historical sandwich/frontrun patterns)
|
|
- DexScreener — pool liquidity, pair metadata
|
|
- Chain RPC (public) — pending tx pool analysis
|
|
- Mempool.space (BTC) and Etherscan pending (EVM)
|
|
- Historical MEV data from mev_sandwich_detector
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from enum import StrEnum
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
__all__ = [
|
|
"MevFactor",
|
|
"MevRiskLevel",
|
|
"MevShieldResult",
|
|
"analyze_transaction_risk",
|
|
"mev_shield_analysis",
|
|
]
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ── Constants ────────────────────────────────────────────────────
|
|
|
|
MEV_API_TIMEOUT = 10
|
|
DEFAULT_SLIPPAGE_BPS = 100 # 1% = 100 bps
|
|
DEFAULT_GAS_PRICE_GWEI = 10
|
|
HIGH_RISK_SLIPPAGE_BPS = 300 # 3%+
|
|
HIGH_RISK_GAS_PRICE_GWEI = 50
|
|
THIN_LIQUIDITY_USD = 50_000
|
|
MODERATE_LIQUIDITY_USD = 250_000
|
|
|
|
FREE_RPCS = {
|
|
"ethereum": "https://eth.llamarpc.com",
|
|
"bsc": "https://bsc-dataseed.binance.org",
|
|
"base": "https://mainnet.base.org",
|
|
"arbitrum": "https://arb1.arbitrum.io/rpc",
|
|
"polygon": "https://polygon-rpc.com",
|
|
"optimism": "https://mainnet.optimism.io",
|
|
"avalanche": "https://api.avax.network/ext/bc/C/rpc",
|
|
}
|
|
|
|
FALLBACK_RPCS = {
|
|
"ethereum": ["https://rpc.ankr.com/eth", "https://ethereum-rpc.publicnode.com"],
|
|
"bsc": ["https://rpc.ankr.com/bsc", "https://bsc-rpc.publicnode.com"],
|
|
"base": ["https://base-rpc.publicnode.com", "https://rpc.base.llama.com"],
|
|
}
|
|
|
|
# ── Risk Levels ──────────────────────────────────────────────────
|
|
|
|
|
|
class MevRiskLevel(StrEnum):
|
|
LOW = "low"
|
|
MODERATE = "moderate"
|
|
HIGH = "high"
|
|
CRITICAL = "critical"
|
|
|
|
|
|
# ── Data Models ──────────────────────────────────────────────────
|
|
|
|
|
|
@dataclass
|
|
class MevFactor:
|
|
name: str
|
|
score: float # 0.0 (safe) to 1.0 (very vulnerable)
|
|
finding: str
|
|
detail: str
|
|
suggestion: str | None = None
|
|
|
|
|
|
@dataclass
|
|
class MevShieldResult:
|
|
risk_level: MevRiskLevel
|
|
overall_score: float # 0.0 (safe) to 1.0 (critical)
|
|
factors: list[MevFactor]
|
|
protection_strategies: list[str]
|
|
estimated_loss_bps: int # Expected loss in basis points if targeted
|
|
timestamp: str
|
|
tx_params: dict[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"risk_level": self.risk_level.value,
|
|
"overall_score": round(self.overall_score, 2),
|
|
"estimated_loss_bps": self.estimated_loss_bps,
|
|
"factors": [
|
|
{
|
|
"name": f.name,
|
|
"score": round(f.score, 2),
|
|
"finding": f.finding,
|
|
"detail": f.detail,
|
|
"suggestion": f.suggestion,
|
|
}
|
|
for f in self.factors
|
|
],
|
|
"protection_strategies": self.protection_strategies,
|
|
"timestamp": self.timestamp,
|
|
"tx_params": self.tx_params,
|
|
}
|
|
|
|
|
|
# ── MEV Shield Analysis Engine ────────────────────────────────────
|
|
async def _fetch_free_rpc(chain: str, method: str, params: list[Any] | None = None) -> dict[str, Any] | None:
|
|
"""Call an RPC with fallback URLs."""
|
|
urls: list[str] = [FREE_RPCS.get(chain, "")]
|
|
if chain in FALLBACK_RPCS:
|
|
urls.extend(FALLBACK_RPCS[chain])
|
|
|
|
for url in urls:
|
|
if not url:
|
|
continue
|
|
try:
|
|
async with httpx.AsyncClient(timeout=MEV_API_TIMEOUT) as client:
|
|
resp = await client.post(
|
|
url,
|
|
json={
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": method,
|
|
"params": params or [],
|
|
},
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
if resp.status_code == 200:
|
|
data: dict[str, Any] = resp.json()
|
|
if "error" not in data:
|
|
return data
|
|
except Exception as e:
|
|
logger.debug(f"RPC {url} failed: {e}")
|
|
continue
|
|
return None
|
|
|
|
|
|
async def _check_mempool_risk(chain: str, pair_address: str | None = None) -> MevFactor:
|
|
"""
|
|
Check pending transaction pool for signs of MEV activity.
|
|
Uses pending tx count and gas price distribution as proxies.
|
|
|
|
Note: pair_address is reserved for future pair-specific mempool analysis.
|
|
Currently performs chain-level mempool congestion assessment only.
|
|
"""
|
|
if chain not in FREE_RPCS:
|
|
return MevFactor(
|
|
name="mempool_activity",
|
|
score=0.3,
|
|
finding=f"Unsupported chain: {chain}",
|
|
detail=f"No RPC endpoints configured for {chain} — mempool analysis unavailable",
|
|
suggestion="Supported chains: " + ", ".join(sorted(FREE_RPCS.keys())),
|
|
)
|
|
score = 0.1
|
|
findings = []
|
|
details = []
|
|
|
|
# Get pending transaction count
|
|
result = await _fetch_free_rpc(chain, "txpool_status")
|
|
if result and "result" in result:
|
|
try:
|
|
pending = int(result["result"].get("pending", "0x0"), 16)
|
|
if pending > 500:
|
|
score = 0.8
|
|
findings.append("Congested mempool")
|
|
details.append(f"{pending} pending transactions — high MEV competition zone")
|
|
elif pending > 200:
|
|
score = 0.5
|
|
findings.append("Moderate mempool activity")
|
|
details.append(f"{pending} pending transactions — MEV possible")
|
|
else:
|
|
score = 0.2
|
|
findings.append("Low mempool activity")
|
|
details.append(f"{pending} pending transactions — lower MEV likelihood")
|
|
except (ValueError, KeyError):
|
|
pass
|
|
|
|
# Check gas prices in pending pool
|
|
result2 = await _fetch_free_rpc(chain, "txpool_content")
|
|
if result2 and "result" in result2:
|
|
try:
|
|
pending_txs = result2["result"].get("pending", {})
|
|
gas_prices = []
|
|
for _nonce_key, txs in pending_txs.items():
|
|
if isinstance(txs, dict):
|
|
for _, tx in txs.items():
|
|
if isinstance(tx, dict) and "gasPrice" in tx:
|
|
gas_prices.append(int(tx["gasPrice"], 16))
|
|
|
|
if gas_prices:
|
|
avg_gas = sum(gas_prices) / len(gas_prices) / 1e9 # convert to gwei
|
|
if avg_gas > 100:
|
|
score = max(score, 0.75)
|
|
findings.append("High gas price environment")
|
|
details.append(f"Avg gas: {avg_gas:.1f} gwei — competitive bidding indicates MEV activity")
|
|
elif avg_gas > 30:
|
|
score = max(score, 0.45)
|
|
findings.append("Elevated gas prices")
|
|
details.append(f"Avg gas: {avg_gas:.1f} gwei — some MEV extraction likely")
|
|
except (ValueError, KeyError):
|
|
pass
|
|
|
|
finding = "; ".join(findings) if findings else "Mempool looks quiet"
|
|
detail = "; ".join(details) if details else "No unusual mempool activity detected"
|
|
suggestions = {
|
|
0.8: "Use private mempool (Flashbots, MEV Blocker) to avoid frontrunning",
|
|
0.5: "Consider using a MEV-protected RPC endpoint",
|
|
0.2: "Standard transaction should be safe — no special protection needed",
|
|
}
|
|
# Pick closest suggestion
|
|
closest_key = min(suggestions.keys(), key=lambda k: abs(k - score))
|
|
suggestion = suggestions[closest_key]
|
|
|
|
return MevFactor(
|
|
name="mempool_activity",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
|
|
|
|
async def _check_slippage_risk(slippage_bps: int) -> MevFactor:
|
|
"""Assess how slippage tolerance exposes the transaction to MEV."""
|
|
if slippage_bps >= HIGH_RISK_SLIPPAGE_BPS:
|
|
score = 0.9
|
|
finding = "High slippage tolerance"
|
|
detail = f"{slippage_bps / 100:.1f}% slippage — easily exploited by sandwich bots"
|
|
suggestion = "Reduce slippage to 0.5-1.0% (50-100 bps) to minimize attack surface"
|
|
elif slippage_bps >= DEFAULT_SLIPPAGE_BPS:
|
|
score = 0.5
|
|
finding = "Moderate slippage tolerance"
|
|
detail = f"{slippage_bps / 100:.1f}% slippage — exploitable in thin liquidity pools"
|
|
suggestion = "Lower slippage to 0.5% for tighter protection"
|
|
else:
|
|
score = 0.15
|
|
finding = "Low slippage tolerance"
|
|
detail = f"{slippage_bps / 100:.1f}% slippage — tight, harder to sandwich"
|
|
suggestion = "Current slippage setting looks safe"
|
|
|
|
return MevFactor(
|
|
name="slippage_tolerance",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
|
|
|
|
async def _check_gas_price_risk(gas_price_gwei: float) -> MevFactor:
|
|
"""Assess if gas price makes the transaction a target."""
|
|
if gas_price_gwei >= HIGH_RISK_GAS_PRICE_GWEI:
|
|
score = 0.7
|
|
finding = "Premium gas price"
|
|
detail = f"{gas_price_gwei:.1f} gwei — high bid flags tx as urgent/value to bots"
|
|
suggestion = "Use Flashbots to submit privately while still getting fast inclusion"
|
|
elif gas_price_gwei >= 20:
|
|
score = 0.4
|
|
finding = "Above-average gas price"
|
|
detail = f"{gas_price_gwei:.1f} gwei — may attract some MEV attention"
|
|
suggestion = "Consider MEV Blocker or a private RPC for extra safety"
|
|
else:
|
|
score = 0.1
|
|
finding = "Normal gas price"
|
|
detail = f"{gas_price_gwei:.1f} gwei — unlikely to attract MEV extractors"
|
|
suggestion = "Standard gas pricing — no special MEV concern"
|
|
|
|
return MevFactor(
|
|
name="gas_price",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
|
|
|
|
async def _check_pool_liquidity(pair_address: str | None, chain: str) -> MevFactor:
|
|
"""Check if pool liquidity is thin enough to enable sandwich attacks."""
|
|
if not pair_address:
|
|
return MevFactor(
|
|
name="pool_liquidity",
|
|
score=0.3,
|
|
finding="Unknown pool",
|
|
detail="No pair address provided — cannot assess liquidity depth",
|
|
suggestion="Provide pair address for accurate MEV risk assessment",
|
|
)
|
|
|
|
try:
|
|
async with httpx.AsyncClient(timeout=MEV_API_TIMEOUT) as client:
|
|
resp = await client.get(f"https://api.dexscreener.com/latest/dex/pairs/{chain}/{pair_address}")
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
pairs = data.get("pairs", [])
|
|
if pairs:
|
|
pair = pairs[0]
|
|
liquidity_usd = float(pair.get("liquidity", {}).get("usd", 0) or 0)
|
|
|
|
if liquidity_usd < THIN_LIQUIDITY_USD:
|
|
score = 0.85
|
|
finding = "Thin liquidity pool"
|
|
detail = (
|
|
f"${liquidity_usd:,.0f} liquidity — highly vulnerable to sandwich/liquidity manipulation"
|
|
)
|
|
suggestion = "Avoid trading in sub-$50K liquidity pools, or use small orders split across time"
|
|
elif liquidity_usd < MODERATE_LIQUIDITY_USD:
|
|
score = 0.5
|
|
finding = "Moderate liquidity"
|
|
detail = f"${liquidity_usd:,.0f} liquidity — sandwich possible for orders >${liquidity_usd * 0.01:,.0f}"
|
|
suggestion = (
|
|
"Keep individual trades under 1% of pool liquidity to minimize slippage and MEV risk"
|
|
)
|
|
else:
|
|
score = 0.15
|
|
finding = "Deep liquidity pool"
|
|
detail = f"${liquidity_usd:,.0f} liquidity — sandwich attacks expensive and unlikely"
|
|
suggestion = "Low MEV risk due to deep liquidity"
|
|
|
|
return MevFactor(
|
|
name="pool_liquidity",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"DexScreener lookup failed: {e}")
|
|
|
|
return MevFactor(
|
|
name="pool_liquidity",
|
|
score=0.3,
|
|
finding="Liquidity lookup failed",
|
|
detail="Could not fetch pool data — conservative risk estimate applied",
|
|
suggestion="Check manually on DexScreener or retry later",
|
|
)
|
|
|
|
|
|
async def _check_historical_mev(chain: str, token_address: str | None = None) -> MevFactor:
|
|
"""
|
|
Check if this chain/token pair has a history of MEV attacks.
|
|
Uses the existing mev_sandwich_detector module if available.
|
|
"""
|
|
score = 0.2
|
|
try:
|
|
# Try to import the MEV detector for historical data
|
|
from app.mev_sandwich_detector import MEVSandwichDetector
|
|
|
|
detector = MEVSandwichDetector()
|
|
stats = await detector.scan()
|
|
if stats and hasattr(stats, "sandwiches"):
|
|
recent_attacks = len(stats.sandwiches)
|
|
if recent_attacks > 20:
|
|
score = 0.75
|
|
finding = "Active MEV chain"
|
|
detail = f"{recent_attacks} MEV attacks in last 24h — high activity chain"
|
|
suggestion = "Always use MEV protection on this chain (Flashbots, MEV Blocker)"
|
|
elif recent_attacks > 5:
|
|
score = 0.45
|
|
finding = "Moderate MEV activity"
|
|
detail = f"{recent_attacks} MEV attacks in last 24h — some risk"
|
|
suggestion = "Consider MEV protection for high-value transactions"
|
|
else:
|
|
score = 0.15
|
|
finding = "Low MEV activity"
|
|
detail = f"{recent_attacks} MEV attacks in last 24h — chain relatively safe"
|
|
suggestion = "Standard transactions should be safe"
|
|
|
|
return MevFactor(
|
|
name="historical_mev",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
except Exception as e:
|
|
logger.debug(f"Historical MEV check failed: {e}")
|
|
|
|
# Fallback: use chain-level heuristics
|
|
chain_risk = {
|
|
"ethereum": 0.6,
|
|
"bsc": 0.5,
|
|
"base": 0.4,
|
|
"arbitrum": 0.35,
|
|
"polygon": 0.3,
|
|
}
|
|
score = chain_risk.get(chain, 0.25)
|
|
return MevFactor(
|
|
name="historical_mev",
|
|
score=round(score, 2),
|
|
finding=f"Default risk for {chain}",
|
|
detail=f"{chain} has {'high' if score > 0.5 else 'moderate' if score > 0.3 else 'low'} historical MEV activity",
|
|
suggestion="Check chain-specific MEV statistics for accurate assessment",
|
|
)
|
|
|
|
|
|
async def _check_timing_risk() -> MevFactor:
|
|
"""Assess if current time is a high-MEV period."""
|
|
now = datetime.now(UTC)
|
|
hour = now.hour
|
|
|
|
# Peak MEV hours tend to be during US market hours (14-22 UTC)
|
|
# and during high-volatility news events
|
|
is_weekend = now.weekday() >= 5
|
|
|
|
if is_weekend:
|
|
score = 0.3
|
|
finding = "Weekend trading"
|
|
detail = "Lower overall MEV activity on weekends — fewer bots competing"
|
|
suggestion = "Good time for lower-risk transactions"
|
|
elif 14 <= hour <= 22:
|
|
score = 0.6
|
|
finding = "Peak MEV hours"
|
|
detail = "US market hours — highest MEV bot competition"
|
|
suggestion = (
|
|
"Consider transacting outside peak hours (before 14:00 UTC) for lower MEV risk, or use private mempool"
|
|
)
|
|
elif 6 <= hour <= 13:
|
|
score = 0.35
|
|
finding = "Off-peak hours"
|
|
detail = "Lower bot activity during Asian/European hours"
|
|
suggestion = "Moderate MEV risk — standard precautions recommended"
|
|
else:
|
|
score = 0.2
|
|
finding = "Low activity period"
|
|
detail = "Nighttime hours — minimal bot activity expected"
|
|
suggestion = "Low MEV risk window"
|
|
|
|
return MevFactor(
|
|
name="timing",
|
|
score=round(score, 2),
|
|
finding=finding,
|
|
detail=detail,
|
|
suggestion=suggestion,
|
|
)
|
|
|
|
|
|
def _compute_protection_strategies(risk_level: MevRiskLevel, factors: list[MevFactor], chain: str) -> list[str]:
|
|
"""Generate actionable protection strategies based on risk profile."""
|
|
strategies = []
|
|
|
|
# Check high-risk factors and suggest specific protections
|
|
high_factors = [f for f in factors if f.score >= 0.6]
|
|
|
|
if risk_level in (MevRiskLevel.CRITICAL, MevRiskLevel.HIGH):
|
|
strategies.append(
|
|
"🚨 Use Flashbots Protect (https://flashbots.net/) for private transaction submission — "
|
|
"guarantees your tx won't be frontrun"
|
|
)
|
|
strategies.append(
|
|
"🔒 Use MEV Blocker (https://mevblocker.io/) — sends tx to private mempool with backrunning protection"
|
|
)
|
|
if any(f.name == "pool_liquidity" for f in high_factors):
|
|
strategies.append("📊 Split large trades into smaller chunks to reduce slippage and sandwich exposure")
|
|
if any(f.name == "slippage_tolerance" for f in high_factors):
|
|
strategies.append(
|
|
"⚡ Set slippage to 0.5% or lower — this is the single most effective MEV reduction tactic"
|
|
)
|
|
elif risk_level == MevRiskLevel.MODERATE:
|
|
strategies.append("🛡️ Consider MEV Blocker for moderate-value transactions — free to use")
|
|
strategies.append("⏰ Time your transaction during off-peak hours (before 14:00 UTC) for lower MEV competition")
|
|
else:
|
|
strategies.append("✅ Standard transaction appears safe — no special MEV protection needed")
|
|
strategies.append("💡 For high-value transactions (>$10K), still consider Flashbots as a precaution")
|
|
|
|
# Chain-specific
|
|
if chain == "ethereum":
|
|
strategies.append(
|
|
"🔷 Ethereum has the most mature MEV protection — Flashbots, MEV Blocker, and CoW Swap all available"
|
|
)
|
|
elif chain == "bsc":
|
|
strategies.append(
|
|
"🟡 BSC has fewer MEV protection options — Flashbots not available. "
|
|
"Use low slippage and consider Poly Network for cross-chain routing"
|
|
)
|
|
|
|
return strategies
|
|
|
|
|
|
def _compute_estimated_loss(risk_level: MevRiskLevel, factors: list[MevFactor]) -> int:
|
|
"""Estimate expected loss in basis points if MEV'd."""
|
|
base_loss = {
|
|
MevRiskLevel.LOW: 10,
|
|
MevRiskLevel.MODERATE: 50,
|
|
MevRiskLevel.HIGH: 150,
|
|
MevRiskLevel.CRITICAL: 300,
|
|
}
|
|
loss = base_loss[risk_level]
|
|
|
|
# Adjust based on specific factors
|
|
for f in factors:
|
|
if f.name == "pool_liquidity" and f.score > 0.7:
|
|
loss = int(loss * 1.5) # Thin liquidity amplifies losses
|
|
if f.name == "slippage_tolerance" and f.score > 0.7:
|
|
loss = int(loss * 1.3) # High slippage enables bigger sandwich
|
|
if f.name == "mempool_activity" and f.score > 0.7:
|
|
loss = int(loss * 1.2) # Competitive mempool = better bots
|
|
|
|
return min(loss, 500) # Cap at 5% max estimated loss
|
|
|
|
|
|
async def _resolve_chain_id(chain: str) -> str:
|
|
"""Normalize chain name to internal format."""
|
|
chain_map = {
|
|
"eth": "ethereum",
|
|
"ether": "ethereum",
|
|
"bsc": "bsc",
|
|
"bnb": "bsc",
|
|
"base": "base",
|
|
"arb": "arbitrum",
|
|
"arbitrum": "arbitrum",
|
|
"polygon": "polygon",
|
|
"matic": "polygon",
|
|
"op": "optimism",
|
|
"optimism": "optimism",
|
|
"avax": "avalanche",
|
|
"avalanche": "avalanche",
|
|
}
|
|
return chain_map.get(chain.lower(), chain.lower())
|
|
|
|
|
|
async def analyze_transaction_risk(
|
|
chain: str = "ethereum",
|
|
pair_address: str | None = None,
|
|
token_address: str | None = None,
|
|
slippage_bps: int | None = None,
|
|
gas_price_gwei: float | None = None,
|
|
) -> MevShieldResult:
|
|
"""
|
|
Full MEV Shield Analysis — assess transaction MEV vulnerability.
|
|
|
|
Args:
|
|
chain: Blockchain name (ethereum, bsc, base, arbitrum, polygon, optimism, avalanche)
|
|
pair_address: DEX pair address for liquidity check (DexScreener format)
|
|
token_address: Token address for historical MEV lookup
|
|
slippage_bps: Slippage tolerance in basis points (default: 100 = 1%)
|
|
gas_price_gwei: Gas price in gwei (default: auto-detect from RPC)
|
|
|
|
Returns:
|
|
MevShieldResult with risk assessment, factor breakdown, and protection strategies
|
|
"""
|
|
chain = await _resolve_chain_id(chain)
|
|
if chain not in FREE_RPCS:
|
|
raise ValueError(f"Unsupported chain: '{chain}'. Supported chains: {', '.join(sorted(FREE_RPCS.keys()))}")
|
|
slippage_bps = slippage_bps or DEFAULT_SLIPPAGE_BPS
|
|
|
|
if gas_price_gwei is None:
|
|
# Auto-detect gas price
|
|
result = await _fetch_free_rpc(chain, "eth_gasPrice")
|
|
if result and "result" in result:
|
|
try:
|
|
gas_price_gwei = int(result["result"], 16) / 1e9
|
|
except (ValueError, KeyError):
|
|
gas_price_gwei = DEFAULT_GAS_PRICE_GWEI
|
|
else:
|
|
gas_price_gwei = DEFAULT_GAS_PRICE_GWEI
|
|
|
|
# Run all risk checks in parallel
|
|
(
|
|
mempool_factor,
|
|
slippage_factor,
|
|
gas_factor,
|
|
liquidity_factor,
|
|
historical_factor,
|
|
timing_factor,
|
|
) = await asyncio.gather(
|
|
_check_mempool_risk(chain, pair_address),
|
|
_check_slippage_risk(slippage_bps),
|
|
_check_gas_price_risk(gas_price_gwei),
|
|
_check_pool_liquidity(pair_address, chain),
|
|
_check_historical_mev(chain, token_address),
|
|
_check_timing_risk(),
|
|
)
|
|
|
|
factors = [
|
|
mempool_factor,
|
|
slippage_factor,
|
|
gas_factor,
|
|
liquidity_factor,
|
|
historical_factor,
|
|
timing_factor,
|
|
]
|
|
|
|
# Compute overall score (weighted average)
|
|
weights = {
|
|
"mempool_activity": 0.20,
|
|
"slippage_tolerance": 0.25,
|
|
"gas_price": 0.15,
|
|
"pool_liquidity": 0.20,
|
|
"historical_mev": 0.10,
|
|
"timing": 0.10,
|
|
}
|
|
|
|
overall_score = sum(f.score * weights.get(f.name, 0.15) for f in factors) / sum(weights.values())
|
|
|
|
# Determine risk level
|
|
if overall_score >= 0.7:
|
|
risk_level = MevRiskLevel.CRITICAL
|
|
elif overall_score >= 0.5:
|
|
risk_level = MevRiskLevel.HIGH
|
|
elif overall_score >= 0.3:
|
|
risk_level = MevRiskLevel.MODERATE
|
|
else:
|
|
risk_level = MevRiskLevel.LOW
|
|
|
|
protection_strategies = _compute_protection_strategies(risk_level, factors, chain)
|
|
estimated_loss_bps = _compute_estimated_loss(risk_level, factors)
|
|
|
|
return MevShieldResult(
|
|
risk_level=risk_level,
|
|
overall_score=round(overall_score, 2),
|
|
factors=factors,
|
|
protection_strategies=protection_strategies,
|
|
estimated_loss_bps=estimated_loss_bps,
|
|
timestamp=datetime.now(UTC).isoformat(),
|
|
tx_params={
|
|
"chain": chain,
|
|
"slippage_bps": slippage_bps,
|
|
"gas_price_gwei": round(gas_price_gwei, 1),
|
|
"pair_address": pair_address,
|
|
"token_address": token_address,
|
|
},
|
|
)
|
|
|
|
|
|
# ── CLI Entry Point ──────────────────────────────────────────────
|
|
|
|
|
|
def mev_shield_analysis(
|
|
chain: str = "ethereum",
|
|
pair_address: str | None = None,
|
|
token_address: str | None = None,
|
|
slippage_bps: int | None = None,
|
|
gas_price_gwei: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Synchronous wrapper for MEV Shield Analysis. Use in non-async contexts."""
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
if loop.is_running():
|
|
# Already in an event loop — create a new one in a new thread
|
|
import concurrent.futures
|
|
|
|
with concurrent.futures.ThreadPoolExecutor() as pool:
|
|
future = pool.submit(
|
|
asyncio.run,
|
|
analyze_transaction_risk(
|
|
chain=chain,
|
|
pair_address=pair_address,
|
|
token_address=token_address,
|
|
slippage_bps=slippage_bps,
|
|
gas_price_gwei=gas_price_gwei,
|
|
),
|
|
)
|
|
return future.result().to_dict()
|
|
except RuntimeError:
|
|
pass
|
|
return asyncio.run(
|
|
analyze_transaction_risk(
|
|
chain=chain,
|
|
pair_address=pair_address,
|
|
token_address=token_address,
|
|
slippage_bps=slippage_bps,
|
|
gas_price_gwei=gas_price_gwei,
|
|
)
|
|
).to_dict()
|
|
|
|
|
|
# ── Main (standalone test) ───────────────────────────────────────
|
|
|
|
|
|
async def main() -> None:
|
|
"""Run a sample MEV Shield analysis for demonstration."""
|
|
print("🛡️ MEV Shield Analysis — Demo Run")
|
|
print("=" * 60)
|
|
|
|
result = await analyze_transaction_risk(
|
|
chain="ethereum",
|
|
pair_address="0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640", # USDC/WETH
|
|
slippage_bps=300, # 3% — high risk
|
|
)
|
|
|
|
print(f"\nRisk Level: {result.risk_level.value.upper()}")
|
|
print(f"Overall Score: {result.overall_score:.2f} / 1.00")
|
|
print(f"Estimated Loss if MEV'd: {result.estimated_loss_bps} bps ({result.estimated_loss_bps / 100:.1f}%)")
|
|
print()
|
|
|
|
print("Factor Breakdown:")
|
|
for f in result.factors:
|
|
bar = "█" * int(f.score * 20) + "░" * (20 - int(f.score * 20))
|
|
print(f" [{bar}] {f.name:25s} {f.score:.2f} — {f.finding}")
|
|
|
|
print("\nProtection Strategies:")
|
|
for s in result.protection_strategies:
|
|
print(f" • {s}")
|
|
|
|
print(f"\nTimestamp: {result.timestamp}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|