איסוף נתוני און-צ'יין: שיטות וכלים

בעת ייצוא היסטוריה מחוקרי בלוקצ'יין, נתקלנו במגבלות חמורות: Etherscan מחזיר לכל היותר 10,000 רשומות לבקשה, 5 בקשות בשנייה בתוכנית החינמית, ללא סטרימינג. אם צריך לאסוף את כל העסקאות של חוזה USDT בשלוש השנים האחרונות – מדובר ביותר מ-30 מיליון רשומות. לניתוח אסטרטגיות DeFi או מעקב אחר עסקאות לווייתנים, נדרשות מיליוני רשומות – ממשקי API סטנדרטיים לא מספיקים. פתרון פשוט לא יעבוד: פגינציה באמצעות page לא תיתן יותר מ-10,000. צריך אסטרטגיה, ומצאנו אותה: פגינציה מבוססת בלוקים עם דדופליקציה לאחר מכן.

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

Alchemy טובה פי 5 מ-Etherscan במגבלות קצב, אין לה מגבלת רשומות בזכות pageKey, תומכת בסטרימינג WebSocket, ועובדת על פני מספר רשתות. לפרויקטים הדורשים ביצועים מקסימליים, Alchemy היא הבחירה המועדפת.

אם צריך לאסוף נתונים היסטוריים לניתוח – צרו קשר, נמצא את הכלי האופטימלי תוך יום אחד. עלויות פרויקט טיפוסיות נעות בין $5,000 ל-$20,000, ולקוחות חוסכים לעיתים קרובות 30-50% על תשתית באמצעות הפתרונות שלנו.

עקיפת מגבלת 10,000 הרשומות של Etherscan API

Etherscan מספק פרמטרים startblock ו-endblock. פגינציה מבוססת בלוקים מאפשרת לשלוף את כל העסקאות. הנה השלבים:

1. הגדר startblock ל-0 (או לבלוק המוקדם ביותר).
2. קרא ל-API עם startblock ו-import httpx import asyncio from typing import AsyncGenerator async def get_all_transactions( address: str, api_key: str, start_block: int = 0 ) -> AsyncGenerator[dict, None]: """Выгрузка ВСЕХ транзакций адреса через пагинацию по блокам""" base_url = "https://api.etherscan.io/api" current_block = start_block while True: async with httpx.AsyncClient() as client: resp = await client.get(base_url, params={ "module": "account", "action": "txlist", "address": address, "startblock": current_block, "endblock": 99999999, "sort": "asc", "apikey": api_key, "offset": 10000, "page": 1, }) data = resp.json() if data["status"] != "1" or not data["result"]: break txs = data["result"] for tx in txs: yield tx if len(txs) < 10000: break current_block = int(txs[-1]["blockNumber"]) + 1 await asyncio.sleep(0.2) .
3. עבד את העסקאות שהוחזרו.
4. אם מספר העסקאות הוא 10000, הגדר startblock למספר הבלוק האחרון + 1 וחזור על שלב 2.
5. אם בלוק בודד מכיל יותר מ-10000 עסקאות, השתמש בפגינציה מקוננת על ידי הגדלת page בתוך אותו בלוק.
6. הוסף השהיה בין בקשות כדי לכבד מגבלות קצב.

הקוד שלהלן עובר על בלוקים, ומגדיל את getAssetTransfers לבלוק האחרון שנתקל בו:

import httpx import asyncio from typing import AsyncGenerator async def get_all_transactions( address: str, api_key: str, start_block: int = 0 ) -> AsyncGenerator[dict, None]: """Выгрузка ВСЕХ транзакций адреса через пагинацию по блокам""" base_url = "https://api.etherscan.io/api" current_block = start_block while True: async with httpx.AsyncClient() as client: resp = await client.get(base_url, params={ "module": "account", "action": "txlist", "address": address, "startblock": current_block, "endblock": 99999999, "sort": "asc", "apikey": api_key, "offset": 10000, "page": 1, }) data = resp.json() if data["status"] != "1" or not data["result"]: break txs = data["result"] for tx in txs: yield tx if len(txs) < 10000: break current_block = int(txs[-1]["blockNumber"]) + 1 await asyncio.sleep(0.2) 

ניואנס חשוב: אם בלוק בודד מכיל >10,000 עסקאות עבור אותה כתובת, הלולאה תיתקע. למקרים כאלה, יש צורך בפגינציה מקוננת עם פרמטר import { Alchemy, Network } from 'alchemy-sdk'; const alchemy = new Alchemy({ apiKey: process.env.ALCHEMY_KEY, network: Network.ETH_MAINNET }); const transfers = await alchemy.core.getAssetTransfers({ fromAddress: '0x...', category: ['external', 'erc20', 'erc721', 'erc1155'], withMetadata: true, maxCount: 1000, }); if (transfers.pageKey) { const more = await alchemy.core.getAssetTransfers({ pageKey: transfers.pageKey, }); } בתוך הבלוק.

היתרונות של Alchemy ו-Moralis על פני Etherscan

Alchemy ו-Moralis מסירות את רוב המגבלות של Etherscan. ל-Alchemy יש מגבלת קצב של 25 בקשות/שנייה, שהיא גבוהה פי 5 מזו של Etherscan (5 בקשות/שנייה). הנה השוואה:

Alchemy getWalletTransactionsVerbose מחזירה ETH + ERC-20 + ERC-721 בקריאה אחת.

import { Alchemy, Network } from 'alchemy-sdk'; const alchemy = new Alchemy({ apiKey: process.env.ALCHEMY_KEY, network: Network.ETH_MAINNET }); const transfers = await alchemy.core.getAssetTransfers({ fromAddress: '0x...', category: ['external', 'erc20', 'erc721', 'erc1155'], withMetadata: true, maxCount: 1000, }); if (transfers.pageKey) { const more = await alchemy.core.getAssetTransfers({ pageKey: transfers.pageKey, }); } 

Alchemy Asset Transfers API ממליצה להשתמש ב-pageKey לפגינציה. Moralis מספקת בנוסף עסקאות פנימיות באמצעות import httpx from bs4 import BeautifulSoup import asyncio async def get_token_holders(token_address: str, pages: int = 10) -> list[dict]: headers = { "User-Agent": "Mozilla/5.0", } holders = [] async with httpx.AsyncClient(headers=headers) as client: for page in range(1, pages + 1): resp = await client.get( f"https://etherscan.io/token/{token_address}", params={"a": "#holders", "p": page} ) soup = BeautifulSoup(resp.text, 'html.parser') table = soup.find('table', {'id': 'holdersTable'}) if not table: break for row in table.find_all('tr')[1:]: cols = row.find_all('td') if len(cols) >= 3: holders.append({ 'rank': cols[0].text.strip(), 'address': cols[1].find('a')['href'].split('/')[-1], 'quantity': cols[2].text.strip(), }) await asyncio.sleep(2) return holders . להערכה מהירה, צרו קשר – נכין אב טיפוס תוך יומיים.

הצדקה לגירוד דפי HTML

אם ה-API לא מחזיר מחזיקי טוקנים או חוזים מאומתים, אנו משתמשים בגירוד HTML. הדוגמה הבאה אוספת מחזיקי טוקנים מ-Etherscan:

import httpx from bs4 import BeautifulSoup import asyncio async def get_token_holders(token_address: str, pages: int = 10) -> list[dict]: headers = { "User-Agent": "Mozilla/5.0", } holders = [] async with httpx.AsyncClient(headers=headers) as client: for page in range(1, pages + 1): resp = await client.get( f"https://etherscan.io/token/{token_address}", params={"a": "#holders", "p": page} ) soup = BeautifulSoup(resp.text, 'html.parser') table = soup.find('table', {'id': 'holdersTable'}) if not table: break for row in table.find_all('tr')[1:]: cols = row.find_all('td') if len(cols) >= 3: holders.append({ 'rank': cols[0].text.strip(), 'address': cols[1].find('a')['href'].split('/')[-1], 'quantity': cols[2].text.strip(), }) await asyncio.sleep(2) return holders 

Etherscan מוגן על ידי Cloudflare – לאיסוף בקנה מידה גדול, יש צורך בפרוקסי מגורים או ב-API הרשמי. גירוד BscScan ו-Solscan עובד באופן דומה.

הימנעות מכפילויות באיסוף מקבילי

בעת איסוף מקבילי ממספר מקורות, כפילויות הן בלתי נמנעות. השתמש ב-ON CONFLICT ב-PostgreSQL:

CREATE TABLE eth_transactions ( tx_hash CHAR(66) PRIMARY KEY, block_number BIGINT NOT NULL, from_address CHAR(42) NOT NULL, to_address CHAR(42), value NUMERIC(38) DEFAULT 0, gas_used BIGINT, status SMALLINT, ts TIMESTAMPTZ ); INSERT INTO eth_transactions VALUES (...) ON CONFLICT (tx_hash) DO NOTHING; 

לאירועים, המפתח הייחודי הוא CREATE TABLE eth_transactions ( tx_hash CHAR(66) PRIMARY KEY, block_number BIGINT NOT NULL, from_address CHAR(42) NOT NULL, to_address CHAR(42), value NUMERIC(38) DEFAULT 0, gas_used BIGINT, status SMALLINT, ts TIMESTAMPTZ ); INSERT INTO eth_transactions VALUES (...) ON CONFLICT (tx_hash) DO NOTHING; .

עבודה ישירה עם נוד

לשלמות מקסימלית (עסקאות פנימיות, משבצות אחסון, MEV), הפעל נוד משלך עם Erigon ו-(tx_hash, log_index). זה נותן:

• כל העסקאות הפנימיות ללא מגבלות
• מעקב אחר קריאות (Call tracing)
• נתוני אחסון

curl -X POST $ETH_RPC_URL \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","method":"eth_getBlockReceipts","params":["0x1234567"],"id":1}' 

--tracing EIP-1559 מחזיר את כל הקבלות של בלוק בבקשה אחת – חלופה ל-N קריאות curl -X POST $ETH_RPC_URL \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","method":"eth_getBlockReceipts","params":["0x1234567"],"id":1}' נפרדות.

שלבי איסוף הנתונים

התהליך שלנו אורך בין 3 ל-10 ימים בהתאם למורכבות.

מה כלול בתוצאה

• תיעוד על סכמת הנתונים ושיטות האיסוף.
• קוד Python/TypeScript מוכן לשימוש עם תמיכה אסינכרונית.
• הגדרת פגינציה לכל חוקר.
• פריסה על שרת או ענן (AWS Lambda, Kubernetes).
• הדרכת צוות על שימוש במערכת.
• חודש תמיכה לאחר המסירה.

הניסיון שלנו מקיף למעלה מ-30 פרויקטים באיסוף וניתוח נתוני on-chain. עלויות פרויקט טיפוסיות מתחילות ב-$5,000 לייצוא פשוט ויכולות להגיע ל-$20,000 להגדרות מורכבות עם ריבוי רשתות. קבלו ייעוץ – נעזור לכם לבחור את הכלי האופטימלי למשימה שלכם. שלחו בקשה, ונקים איסוף נתונים בכל מורכבות.