ניתוח טוקנים חדשים: זיהוי, ניתוח וסיכונים

אתה מפעיל בוט ל-MEV או לסריקת טוקנים חדשים ומתמודד עם בעיה: איך לזהות חוזים בצורה אמינה בזמן אמת? אנחנו פותרים משימה זו באמצעות מספר מנגנונים מקבילים — מניטור חוזי factory ועד ניתוח עסקאות deployment. לצוות שלנו יש ניסיון של 5+ שנים בפיתוח blockchain והוא יישם סורקים ל-15+ פרויקטים ב-DeFi. אנו מבטיחים פעילות יציבה וזמן השהיה נמוך, אפילו בעומסי שיא: Ethereum מייצר עד 15 בלוקים לדקה, כל אחד עם מאות עסקאות.

אחד הלקוחות שלנו, קרן גידור, השתמש בסורק לזיהוי מוקדם של טוקנים ב-Arbitrum והשיג הפחתת זמן השהיה מ-15 שניות ל-2. זה אפשר להם להיות ראשונים בבריכות נזילות, והגדיל את ה-ROI ב-30% לרבעון.

איך אנחנו מנתחים נתוני טוקנים חדשים

איך לזהות טוקן חדש On-Chain?

שיטה 1: ניטור חוזי Factory

רוב הטוקנים מושקים דרך factories: Uniswap V2/V3 factory בעת יצירת בריכה, חוזי Token Factory, או deployment ישיר עם אירוע. Uniswap V2 factory פולט PairCreated כאשר נוצרת בריכה חדשה — זהו האות האמין ביותר לטוקן חדש שניתן לסחור בו:

from web3 import Web3 import asyncio UNISWAP_V2_FACTORY = "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f" PAIR_CREATED_TOPIC = "0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9" FACTORY_ABI = [{ "name": "PairCreated", "type": "event", "inputs": [ {"name": "token0", "type": "address", "indexed": True}, {"name": "token1", "type": "address", "indexed": True}, {"name": "pair", "type": "address", "indexed": False}, {"name": "", "type": "uint256", "indexed": False} ] }] async def watch_new_pairs(w3: Web3, callback): factory = w3.eth.contract(address=UNISWAP_V2_FACTORY, abi=FACTORY_ABI) # Подписка через WebSocket на новые события event_filter = await w3.eth.filter({ "address": UNISWAP_V2_FACTORY, "topics": [PAIR_CREATED_TOPIC] }) while True: events = await event_filter.get_new_entries() for event_log in events: decoded = factory.events.PairCreated().process_log(event_log) await callback({ "token0": decoded.args.token0, "token1": decoded.args.token1, "pair": decoded.args.pair, "block": event_log.blockNumber, "tx_hash": event_log.transactionHash.hex() }) await asyncio.sleep(3) 

עבור Uniswap V3 — באופן דומה, האזן ל-from web3 import Web3 import asyncio UNISWAP_V2_FACTORY = "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f" PAIR_CREATED_TOPIC = "0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9" FACTORY_ABI = [{ "name": "PairCreated", "type": "event", "inputs": [ {"name": "token0", "type": "address", "indexed": True}, {"name": "token1", "type": "address", "indexed": True}, {"name": "pair", "type": "address", "indexed": False}, {"name": "", "type": "uint256", "indexed": False} ] }] async def watch_new_pairs(w3: Web3, callback): factory = w3.eth.contract(address=UNISWAP_V2_FACTORY, abi=FACTORY_ABI) # Подписка через WebSocket на новые события event_filter = await w3.eth.filter({ "address": UNISWAP_V2_FACTORY, "topics": [PAIR_CREATED_TOPIC] }) while True: events = await event_filter.get_new_entries() for event_log in events: decoded = factory.events.PairCreated().process_log(event_log) await callback({ "token0": decoded.args.token0, "token1": decoded.args.token1, "pair": decoded.args.pair, "block": event_log.blockNumber, "tx_hash": event_log.transactionHash.hex() }) await asyncio.sleep(3) על PoolCreated.

שיטה 2: זיהוי deployment של חוזה ERC-20

deployment ישיר של ERC-20 אינו פולט אירועים סטנדרטיים. אנחנו מזהים זאת על ידי ניתוח קבלות עסקאות — אם 0x1F98431c8aD98523631AE4a59f267346ea31F984 אינו ריק, מדובר ב-deployment של חוזה:

async def scan_block_for_deployments(block_number: int, w3: Web3) -> list[dict]: block = w3.eth.get_block(block_number, full_transactions=True) deployments = [] for tx in block.transactions: if tx.to is None: # tx без to = деплой контракта receipt = w3.eth.get_transaction_receipt(tx.hash) if receipt.contractAddress: # Проверяем, является ли ERC-20 token_info = await check_if_erc20(receipt.contractAddress, w3) if token_info: deployments.append({ "contract": receipt.contractAddress, "deployer": tx["from"], "block": block_number, "tx_hash": tx.hash.hex(), **token_info }) return deployments async def check_if_erc20(address: str, w3: Web3) -> dict | None: """Проверяем наличие обязательных ERC-20 методов""" minimal_abi = [ {"name": "totalSupply", "type": "function", "inputs": [], "outputs": [{"type": "uint256"}]}, {"name": "decimals", "type": "function", "inputs": [], "outputs": [{"type": "uint8"}]}, {"name": "symbol", "type": "function", "inputs": [], "outputs": [{"type": "string"}]}, {"name": "name", "type": "function", "inputs": [], "outputs": [{"type": "string"}]}, ] try: contract = w3.eth.contract(address=address, abil=minimal_abi) return { "name": contract.functions.name().call(), "symbol": contract.functions.symbol().call(), "decimals": contract.functions.decimals().call(), "total_supply": contract.functions.totalSupply().call() } except Exception: return None # не ERC-20 или reverting контракт 

סריקת כל בלוק היא עומס גבוה על RPC. ברשת הראשית של Ethereum ~15 בלוקים/דקה, כל אחד עשוי להכיל מאות עסקאות. אתה צריך תוכנית ייעודית של Alchemy/QuickNode או צומת משלך. ניטור factory עובד פי 10 מהר יותר מסריקת בלוקים מלאה.

העשרת נתונים לאחר זיהוי

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

async def enrich_new_token(contract_address: str, w3: Web3) -> dict: tasks = await asyncio.gather( get_contract_source_code(contract_address), # Etherscan API get_lp_info(contract_address), # существующие пулы get_social_links(contract_address), # из контракта или Etherscan run_honeypot_check(contract_address), # налог на продажу, торгуемость return_exceptions=True ) source, lp_info, socials, honeypot = tasks return { "verified_source": bool(source and not isinstance(source, Exception)), "has_liquidity": bool(lp_info and not isinstance(lp_info, Exception)), "honeypot_risk": honeypot if not isinstance(honeypot, Exception) else "unknown", **socials if not isinstance(socials, Exception) else {} } 

ניתוח סיכונים אוטומטי של טוקנים

עבור סורק עם התראות — ניתוח אוטומטי של bytecode והתנהגות:

SCAM_PATTERNS = { "mint_function": "0x40c10f19", # bytes4 selector для mint(address, uint256) "ownership_transfer": "0xf2fde38b", "blacklist_function": "0x44337ea1", } def check_bytecode_risks(bytecode: str) -> list[str]: risks = [] if len(bytecode) < 100: risks.append("minimal_bytecode") # прокси или пустышка for name, selector in SCAM_PATTERNS.items(): if selector[2:] in bytecode: # убираем 0x risks.append(name) return risks 

בדיקת honeypot אמיתית דורשת סימולציה של עסקאות קנייה ומכירה באמצעות contractAddress — זה קובע מס קנייה/מכירה והאם ניתן בכלל למכור את הטוקן. שירותים: honeypot.is API, GoPlus Security API.

אחסון ואינדוקס

CREATE TABLE new_tokens ( id BIGSERIAL PRIMARY KEY, chain_id INTEGER NOT NULL, contract TEXT NOT NULL, name TEXT, symbol TEXT, decimals SMALLINT, total_supply NUMERIC, deployer TEXT NOT NULL, deploy_block INTEGER NOT NULL, deploy_tx TEXT NOT NULL, deploy_time TIMESTAMPTZ NOT NULL, verified BOOLEAN DEFAULT FALSE, has_liquidity BOOLEAN DEFAULT FALSE, risk_flags TEXT[] DEFAULT '{}', enriched_at TIMESTAMPTZ, UNIQUE(chain_id, contract) ); CREATE INDEX ON new_tokens(deploy_time DESC); CREATE INDEX ON new_tokens(chain_id, symbol); CREATE INDEX ON new_tokens USING gin(risk_flags); 

אילו סיכונים הסורק מזהה?

אנחנו מסווגים סיכונים לשלוש רמות: קריטי (honeypot, reentrancy), גבוה (mint, blacklist), ובינוני (עמלה גבוהה, נזילות נמוכה). עבור כל טוקן נוצר סיכום עם דגלים והמלצות. זה מאפשר דחייה מיידית של 90% מטוקני ההונאה בשלב הזיהוי.

השוואת שיטות זיהוי

למה ניטור Factory מהיר יותר מסריקת בלוקים

חוזי factory פולטים אירוע כאשר נוצרת בריכה — זהו האות היחיד שצריך לעקוב אחריו. סריקת בלוקים מלאה דורשת עיבוד של כל עסקה, מה שמגדיל את עומס ה-RPC פי 10-15 ומאט את הזיהוי. לכן, עבור מערכות production, אנו ממליצים לשלב את שתי השיטות אך לתעדף אירועי factory.

מה כלול בפיתוח סורק

התהליך שלנו

1. אנליטיקה — אנו לומדים את הדרישות שלך, בוחרים רשתות ושיטות.
2. עיצוב — יוצרים ארכיטקטורה וסכימת נתונים.
3. יישום — כותבים קוד לניתוח, העשרה ו-API.
4. בדיקות — מדמים עומס, מאמתים על נתונים אמיתיים.
5. Deployment — פורסים על התשתית שלך או שלנו.
6. תמיכה — מאמנים את הצוות שלך, מעבירים תיעוד.

לוח זמנים ועלות

זמן פיתוח לסורק בסיסי ל-3-4 רשתות EVM: בין 3 ל-5 שבועות. העלות מחושבת באופן אישי לפי מורכבות. קבל ייעוץ — נעריך את הפרויקט שלך ונציע פתרון אופטימלי. הזמן פיתוח סורק לזיהוי מוקדם של טוקנים והפחת את עלויות הניטור.