Parsing New Tokens: Detection, Analysis, and Risks

You launch a bot for MEV or screening new tokens and face a problem: how to reliably detect contracts in real time? We solve this task through several parallel mechanisms — from monitoring factory contracts to analyzing deployment transactions. Our team has 5+ years of experience in blockchain development and has implemented scanners for 15+ projects in DeFi. We guarantee stable operation and low latency, even at peak loads: Ethereum generates up to 15 blocks per minute, each with hundreds of transactions.

One of our clients, a hedge fund, used the scanner for early token detection on Arbitrum and achieved a latency reduction from 15 seconds to 2. This allowed them to be first into liquidity pools, increasing ROI by 30% per quarter.

How We Parse New Token Data

How to Detect a New Token On-Chain?

Method 1: Monitoring Factory Contracts

Most tokens are deployed through factories: Uniswap V2/V3 factory when creating a pool, Token Factory contracts, or direct deployment with an event. Uniswap V2 factory emits PairCreated when a new pool is created — this is the most reliable signal of a new tradable token:

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) # Subscribe via WebSocket to new events 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) 

For Uniswap V3 — similarly, listen for PoolCreated on 0x1F98431c8aD98523631AE4a59f267346ea31F984.

Method 2: Detecting ERC-20 Contract Deployment

Direct ERC-20 deployment does not emit standard events. We detect it by analyzing transaction receipts — if contractAddress is non-empty, it's a contract 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 without to = contract deployment receipt = w3.eth.get_transaction_receipt(tx.hash) if receipt.contractAddress: # Check if it's 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: """Check for mandatory ERC-20 methods""" 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 # not ERC-20 or reverting contract 

Scanning every block is high load on RPC. On Ethereum mainnet ~15 blocks/minute, each may have hundreds of transactions. You need a dedicated Alchemy/QuickNode plan or your own node. Factory monitoring works 10 times faster than full block scanning.

Data Enrichment After Detection

A bare contract address is not very informative. Immediately after detection, we enrich:

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), # existing pools get_social_links(contract_address), # from contract or Etherscan run_honeypot_check(contract_address), # sell tax, tradability 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 {} } 

Automatic Token Risk Analysis

For a scanner with alerts — automatic bytecode and behavior analysis:

SCAM_PATTERNS = { "mint_function": "0x40c10f19", # bytes4 selector for 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") # proxy or placeholder for name, selector in SCAM_PATTERNS.items(): if selector[2:] in bytecode: # remove 0x risks.append(name) return risks 

A real honeypot check requires simulating buy and sell transactions via eth_call — this determines buy/sell tax and whether you can sell the token at all. Services: honeypot.is API, GoPlus Security API.

Storage and Indexing

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); 

What Risks Does the Scanner Detect?

We categorize risks into three levels: critical (honeypot, reentrancy), high (mint, blacklist), and medium (high fee, low liquidity). For each token, a summary with flags and recommendations is generated. This allows immediate rejection of 90% of scam tokens at the detection stage.

Comparison of Detection Methods

Why Factory Monitoring Is Faster Than Block Scanning

Factory contracts emit an event when a pool is created — this is the only signal that needs to be tracked. Full block scanning requires processing every transaction, increasing RPC load by 10-15 times and slowing detection. Therefore, for production systems, we recommend combining both methods but prioritizing factory events.

What's Included in Scanner Development

Our Process

1. Analytics — we study your requirements, select networks and methods.
2. Design — create architecture and data schema.
3. Implementation — write parsing, enrichment, and API code.
4. Testing — simulate load, verify on real data.
5. Deployment — deploy on your infrastructure or ours.
6. Support — train your team, hand over documentation.

Timeline and Cost

Development time for a basic scanner for 3-4 EVM networks: from 3 to 5 weeks. Cost is calculated individually based on complexity. Get a consultation — we will evaluate your project and offer an optimal solution. Order scanner development for early token detection and reduce monitoring costs.