ניתוח Mempool ותשתית MEV: איסוף נתונים עם זמן השהיה נמוך

כל צומת Ethereum מלא מנהל txpool—מבנה בזיכרון של עסקאות לא מאושרות. שיטת ה-RPC txpool_content מחזירה את כל ה-pool, אך היא שאילתה כבדה. המנוי ב-WebSocket eth_subscribe("pendingTransactions") מספק זרם של hashes אך דורש בקשה נפרדת לפרטים. הארכיטקטורה שלנו משתמשת בשילוב של שיטות למהירות מרבית.

ה-mempool אינו גלובלי. בשל טופולוגיית P2P, צמתים שונים רואים קבוצות שונות. עבור יישומים רגישים ל-MEV, חשוב להתחשב ב-mempool פרטיים.

eth_subscribe עם גוף עסקה מלא

השיטה היעילה ביותר היא מנוי WebSocket עם הדגל true כדי לכלול את הגוף המלא:

import asyncio import json import websockets async def subscribe_mempool_full(): async with websockets.connect("wss://your-private-node: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) 

לא כל הספקים תומכים ב-import asyncio import json import websockets async def subscribe_mempool_full(): async with websockets.connect("wss://your-private-node: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) . Alchemy ו-Infura תומכים, אך RPC ציבורי מגביל מנויים.

txpool_content עבור תמונות מצב

לתמונת מצב מלאה של ה-mempool, אנו משתמשים ב-True—רק על צומת פרטי. שיטה זו מאפשרת ניתוח txpool בכל עומק:

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 

בקשה זו כבדה—לא יותר מפעם בשנייה.

מדוע צומת פרטי הוא קריטי עבור MEV

RPC ציבורי (Alchemy, Infura) מגביל מנויים ממתינים ואינו מאפשר txpool_content. רק צומת פרטי מאפשר:

• קבלת עסקאות עם זמן השהיה מינימלי (רמת P2P)
• שימוש ב-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 ללא הגבלות
• חיבור מופעים נוספים לסובלנות תקלות

אנו פורסים צמתים על Geth/Reth עם 32 GB RAM ו-NVMe SSDs. ההשקעה בצומת פרטי מחזירה את עצמה תוך חודשיים בפעילות בוט ממוצעת.

פענוח Calldata בזמן אמת

4 הבתים הראשונים של calldata הם בורר הפונקציות. זה מזהה את שיטת הפרוטוקול הנקראת. אנו משתמשים במסד הנתונים 4byte.directory וב-ABIs טעונים. הפענוח מהיר פי 3–5 מחלופות בשל חתימות מאוחסנות מראש:

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} 

עבור בוררים לא ידועים, אנו שואלים את ה-API של 4byte.

ארכיטקטורת Monitor בעלת ביצועים גבוהים

[Собственные ноды] → [Kafka: raw tx stream] ↓ [Decoder Worker Pool] / | \ [MEV Detector] [Volume Monitor] [Alert Engine] ↓ [TimescaleDB / ClickHouse] 

מה כלול

• קוד אינטגרציה מלא (Python / Rust / TypeScript)
• תיעוד API עם דוגמאות בקשות
• לוח מחוונים מטרי (Grafana) עם ויזואליזציה של זמן השהיה ותפוקה
• 30 ימי תמיכה טכנית לאחר ההטמעה
• ייעוץ אופטימיזציה לארכיטקטורה עבור הפרויקט שלך

שלבי הטמעה:

1. בחירת טופולוגיית צמתים (Ethereum, Solana וכו')
2. הגדרת Kafka/Redis Streams
3. פענוח והעשרת עסקאות
4. זיהוי דפוסי MEV (sandwich, ארביטראז', frontrunning)
5. הגדרת לוח מחוונים והתראות

מאפיינים של רשתות אחרות

זיהוי דפוסי MEV

בהתבסס על נתוני mempool, אנו מזהים:

• התקפות Sandwich: החלפה גדולה המוקפת בשתי עסקאות מנוגדות
• ארביטראז': פערי מחירים בין DEX שונים
• Front-running: עסקאות עם גז גבוה המעתיקות אסטרטגיות ידועות

דוגמה לזיהוי Sandwich:

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 

הגלאי שלנו מעבד עד 300 עסקאות בשנייה עם דיוק של 95%.

ניטור ואחסון

מדדים מרכזיים:

• פיגור Mempool (יעד: <100 אלפיות שנייה)
• תפוקת מפענח (חייבת לכסות את הזרם הנכנס)
• הודעות שנפלו (0% אובדן ב-Kafka)
• מספר עסקאות ממתינות (חריגה >200K = עומס)

מדיניות שמירה:

כיצד אנו מבטיחים איכות נתונים

אנו משתמשים במספר רמות אימות: בדיקת כפילויות, הצלבה עם בלוק האישור, ניטור אובדן. אם מתגלות חריגות, המערכת עוברת אוטומטית לצומת גיבוי. זה ממזער זמן השבתה ואובדן נתונים.

כדי להגן מפני התקפות sandwich ו-frontrunning, אנו מיישמים סינון של עסקאות עם פרמטרים חשודים. כל הפתרונות מותאמים לאסטרטגיה שלך.

צור קשר להערכת פרויקט וארכיטקטורה סוהרת. ההטמעה אורכת מ-5 ימי עסקים.