# Token Nft Scam Investigation > Use when investigate token and NFT scams including rug pulls, honeypot tokens, pump-and-dump schemes, wash trading, and NFT floor manipulation to identify fraudulent patterns and trace perpetrator wallets. Use when analyzing suspicious token launches, investigating NFT fraud, or detecting market manipulation. - Skill: `oyi77/token-nft-scam-investigation` (Agent Skill) - Install (CLI): `npx skillmds add oyi77/token-nft-scam-investigation` - Raw SKILL.md: https://api.skillmd.com/api/skills/oyi77/token-nft-scam-investigation/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: oyi77 (https://skillmd.com/u/oyi77) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/oyi77/token-nft-scam-investigation --- # Token & NFT Scam Investigation ## Overview Token and NFT scams — including rug pulls, honeypot tokens, pump-and-dump schemes, wash trading, and floor manipulation — account for the majority of blockchain fraud by victim count. Unlike protocol exploits that target code flaws, these scams target retail investors through deception and market manipulation. This skill covers analyzing token contracts for honeypot mechanisms, detecting wash trading patterns on NFT marketplaces, investigating rug pull setups (liquidity removal, mint controls, proxy upgrades), identifying pump-and-dump coordination, and tracing scam proceeds. It provides working Python code for each detection category, real case studies with exploitable contract addresses, and a monetization framework for selling investigation services. ## When to Use **Trigger phrases:** - "token scam investigation" - "nft scam investigation" - "Analyze a rug pull" - "Check if a token is a honeypot" - "Detect wash trading in an NFT collection" - "is this token safe to buy" - "check token contract for honeypot" - "investigate NFT wash trading" - "trace rug pull proceeds" - "check if this is a scam token" - "due diligence on token launch" - "verify liquidity lock" - When investigating a suspicious token launch for potential scam indicators - When analyzing an NFT collection for wash trading or price manipulation - When tracing rug pull or pump-and-dump proceeds - When building automated scam detection for token/NFT due diligence - When gathering evidence for a fraud complaint or bounty submission - When performing pre-purchase due diligence on a new token or NFT collection - When investigating a project that claims to have been "rugged" for insurance claims - When conducting competitive intelligence on scam operations ## When NOT to Use - You need to trace stolen funds from a hack or exploit (use onchain-transaction-forensics skill) - You need to analyze a DeFi protocol's smart contract vulnerability (use defi-incident-analysis skill) - Task requires recovering lost funds or filing a police report (use proper legal channels) - You are the project owner needing to prove your token is legitimate (use a different due diligence process) - The investigation requires access to private sale or KYC data you don't have - You need to audit a complex DeFi protocol's tokenomics model (use smart-contract-exploiter skill) - You are building a trading bot that needs MEV protection analysis (use onchain-transaction-forensics skill) - The token is a well-known blue-chip project and you're looking for FUD evidence ## Prerequisites - Python 3.8+ with web3.py (>=6.0), requests, pandas, numpy, networkx - Ethereum RPC node access (Alchemy, Infura, or local node) or chain-specific node (BSC, Polygon, etc.) - Block explorer API key (Etherscan, BscScan, Polygonscan) for historical event queries - Solidity understanding (for contract-level scam analysis including proxy patterns and fee mechanisms) - NFT marketplace API access (OpenSea, Blur, LooksRare) for trading data - Familiarity with DEX concepts (Uniswap v2/v3, PancakeSwap, LP tokens) - git (for cloning known scam contract source code from audit repositories) ## Core Workflow ```python # Entry point: run all scam checks against a token contract from web3 import Web3 from typing import Optional, Dict, Any, List, Tuple from dataclasses import dataclass, field import json import time import requests # Configuration ETHERSCAN_API_KEY = "YOUR_API_KEY" RPC_URL = "https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY" w3 = Web3(Web3.HTTPProvider(RPC_URL)) # Standard ERC-20 ABI fragments ERC20_ABI = [ {"inputs": [], "name": "name", "outputs": [{"type": "string"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "symbol", "outputs": [{"type": "string"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "decimals", "outputs": [{"type": "uint8"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "totalSupply", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"type": "address"}], "name": "balanceOf", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"type": "address"}, {"type": "address"}], "name": "allowance", "outputs": [{"type": "uint256"}], "stateMutability": "view", "type": "function"}, # Transfer event {"anonymous": False, "inputs": [{"indexed": True, "name": "from", "type": "address"}, {"indexed": True, "name": "to", "type": "address"}, {"indexed": False, "name": "value", "type": "uint256"}], "name": "Transfer", "type": "event"}, ] # Ownership ABIs to test OWNERSHIP_ABIS = { "ownable": [ {"inputs": [], "name": "owner", "outputs": [{"type": "address"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "renounceOwnership", "outputs": [], "stateMutability": "nonpayable", "type": "function"}, ], "access_control": [ {"inputs": [{"type": "bytes32"}], "name": "hasRole", "outputs": [{"type": "bool"}], "stateMutability": "view", "type": "function"}, {"inputs": [], "name": "DEFAULT_ADMIN_ROLE", "outputs": [{"type": "bytes32"}], "stateMutability": "view", "type": "function"}, ], } # ERC-1967 proxy storage slots (beige paper) ERC1967_IMPLEMENTATION_SLOT = "0x360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc" ERC1967_BEACON_SLOT = "0xa3f0ad74e5423aebfd80d3ef4346578335a9a72aeaee59ff6cb3582b35133d50" ADMIN_SLOT = "0xb53127684a568b3173ae13b9f8a6016e243e63b6e8ee1178d6a717850b5d6103" def run_scam_check(address: str, rpc_url: str = RPC_URL) -> dict: """Run a comprehensive scam check on a token contract.""" w3_local = Web3(Web3.HTTPProvider(rpc_url)) checks = { "basic_info": check_basic_info(w3_local, address), "honeypot": simulate_honeypot(w3_local, address), "proxy": check_proxy_upgradeability(w3_local, address), "ownership": check_all_ownership_interfaces(w3_local, address), "holder_distribution": analyze_holder_distribution(w3_local, address), "approval_risks": check_approval_abuse(w3_local, address), } return checks ``` ### Step 1: Token Contract Analysis Examine the token contract for scam indicators: - **Honeypot detection** — can the token be bought but not sold? Check for blacklist functions, transfer fee manipulation, or balance restrictions - **Ownership not renounced** — deployer can still mint, pause transfers, or blacklist addresses - **Suspicious fee structure** — extremely high buy/sell taxes that benefit the deployer - **Honeypot simulation** — simulate a buy and sell transaction to verify both succeed - **Proxy/upgradeable pattern** — contract can be upgraded to malicious logic later - **Mint function** — unlimited mint capability that dilutes holders - **TOP10 holder concentration** — if 90%+ supply is in a few addresses, risk of dump ```python def check_basic_info(w3: Web3, token_address: str) -> dict: """Fetch basic ERC-20 token metadata and catch common ABI failures.""" contract = w3.eth.contract(address=token_address, abi=ERC20_ABI) result = {} for field in ["name", "symbol", "decimals", "totalSupply"]: try: fn = getattr(contract.functions, field) result[field] = fn().call() except Exception as e: result[field] = f"ERROR: {e}" return result def check_all_ownership_interfaces(w3: Web3, token_address: str) -> dict: """Probe multiple ownership interfaces and check renounced status.""" results = {"interfaces_found": []} # 1. Ownable (OpenZeppelin standard) try: o_contract = w3.eth.contract(address=token_address, abi=OWNERSHIP_ABIS["ownable"]) owner = o_contract.functions.owner().call() is_renounced = owner == "0x0000000000000000000000000000000000000000" or owner == "0x0000000000000000000000000000000000000001" results["ownable_owner"] = owner results["ownable_renounced"] = is_renounced results["interfaces_found"].append("Ownable") except Exception: results["ownable_owner"] = None results["ownable_renounced"] = None # 2. AccessControl (OpenZeppelin v4+) try: ac_contract = w3.eth.contract(address=token_address, abi=OWNERSHIP_ABIS["access_control"]) admin_role = ac_contract.functions.DEFAULT_ADMIN_ROLE().call() # Try to find the admin role holder via event log results["access_control_found"] = True results["default_admin_role"] = admin_role.hex() results["interfaces_found"].append("AccessControl") except Exception: results["access_control_found"] = False # 3. Check if deployer still has a special "minter" style role deployer = _get_contract_deployer(w3, token_address) if deployer and results.get("ownable_owner"): results["deployer_is_owner"] = deployer.lower() == results["ownable_owner"].lower() # 4. Look for management functions via signature detection management_sigs = { "0x8456cb59": "pause()", "0x3f4ba83a": "unpause()", "0x42966c68": "burn(address,uint256)", "0x40c10f19": "mint(address,uint256)", "0x9dc29fac": "blacklist(address)", "0xe0021f43": "setBlacklist(address,bool)", "0x8da5cb5b": "owner()", } detected_sigs = _detect_function_signatures(w3, token_address, list(management_sigs.keys())) results["management_functions"] = [management_sigs[s] for s in detected_sigs if s in management_sigs] return results def _get_contract_deployer(w3: Web3, address: str) -> Optional[str]: """Get deployer address from the creation transaction.""" try: # Get creation transaction by scanning block explorer-style: # Check the contract creation tx via eth_getTransactionReceipt # The deployer is the 'from' field of the tx that created this contract. # For Archive Nodes: scan creation via eth_getTransactionByHash from a known deploy tx. # For non-archive: use block explorer API or internal tx scanning. # Approach: get code, then scan recent blocks for the CREATE opcode caller. checksum_addr = Web3.to_checksum_address(address) # Strategy: check each block for a ContractCreated log or scan deployer list # Since most RPCs don't have debug_traceTransaction, fall back to # checking if the deployer can be inferred from the first Transfer event. transfer_sig = w3.keccak(text="Transfer(address,address,uint256)").hex() latest = w3.eth.block_number # Scan last 2000 blocks from genesis area for old tokens, or recent for new ones # A token created in last 100k blocks = new. Beyond that = can't trace without archive node. logs = w3.eth.get_logs({ "address": checksum_addr, "fromBlock": 0, "toBlock": 1, # Very first block — if token is old, this fails gracefully "topics": [transfer_sig], }) # If we got here without error, token is old and we can't trace deployer easily return None except Exception as e: # Most RPCs will error on too-wide block range — fallback to empty return None def _detect_function_signatures(w3: Web3, address: str, sigs: List[str]) -> List[str]: """Detect which function signatures exist in contract bytecode.""" code = w3.eth.get_code(Web3.to_checksum_address(address)) if code.hex() == "0x": return [] found = [] code_hex = code.hex() for sig in sigs: if sig[2:] in code_hex: found.append(sig) return found ``` ### Step 2: Liquidity Pool Investigation Analyze the token's liquidity on DEXes: - **Liquidity lock status** — is LP locked, and if so, for how long? - **LP holder concentration** — who holds the LP tokens? Is it the deployer? - **Initial liquidity amount** — was seeding adequate for the market cap? - **Sniper activity** — was there bot-buying in the first block after launch? ## Full Honeypot Simulation The core test for a honeypot token: simulate a buy transaction (swap 0.1 ETH for tokens on a DEX), then simulate a sell of the received tokens back to ETH. If the sell transaction fails gas estimation or reverts, the token is a honeypot. ```python def simulate_honeypot( w3: Web3, token_address: str, router_address: str = "0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D", # Uniswap V2 Router amount_in_eth: float = 0.01, from_address: str = "0x000000000000000000000000000000000000dEaD", ) -> dict: """ Simulate a buy and immediate sell on a DEX to detect honeypot tokens. Uses eth_call (no gas spent) to test both directions. Returns dict with buy_success, sell_success, and error details. """ token = w3.eth.contract(address=Web3.to_checksum_address(token_address), abi=ERC20_ABI) router = w3.eth.contract( address=Web3.to_checksum_address(router_address), abi=[{"inputs": [{"internalType": "uint256", "name": "amountIn", "type": "uint256"}, {"internalType": "address[]", "name": "path", "type": "address[]"}], "name": "getAmountsOut", "outputs": [{"internalType": "uint256[]", "name": "amounts", "type": "uint256[]"}], "stateMutability": "view", "type": "function"}], ) weth_address = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2" result = {"buy_success": False, "sell_success": False, "is_honeypot": None, "details": {}} try: # Step 1: Simulate buy (WETH -> Token) amount_in_wei = w3.to_wei(amount_in_eth, "ether") path_buy = [weth_address, token_address] amounts_out_buy = router.functions.getAmountsOut(amount_in_wei, path_buy).call() tokens_received = amounts_out_buy[1] result["buy_success"] = True result["details"]["tokens_received"] = tokens_received result["details"]["buy_price_per_token"] = amount_in_wei / tokens_received if tokens_received > 0 else float("inf") # Check if buy succeeds but gives nearly nothing (tax trap) if tokens_received < amount_in_wei / 1000: # less than 0.1% expected result["details"]["buy_warning"] = "Extremely low token output — possible high buy tax or manipulated price" # Step 2: Simulate sell (Token -> WETH) — the honeypot test path_sell = [token_address, weth_address] sell_amount = tokens_received min_sell = max(1, sell_amount // 1000) # tolerate up to 99.9% sell tax amounts_out_sell = router.functions.getAmountsOut(sell_amount, path_sell).call() eth_back = amounts_out_sell[1] result["sell_success"] = True result["details"]["eth_back"] = eth_back result["details"]["sell_tax_pct"] = max(0.0, (1 - eth_back / amount_in_wei) * 100) result["details"]["is_profitable"] = eth_back > amount_in_wei * 0.5 # get back more than 50% of initial # Extreme sell tax check if result["details"]["sell_tax_pct"] > 50: result["details"]["sell_warning"] = f"Honeypot indicator: sell tax is {result['details']['sell_tax_pct']:.1f}%" # Step 3: Check for blacklist by testing a known blacklisted address # Some honeypots block all addresses except whitelisted ones result["details"]["blacklist_test"] = _check_blacklist_behavior(w3, token_address) # Step 4: Check if minter can deploy infinite supply result["details"]["mint_test"] = _check_mint_function(w3, token_address) except Exception as e: error_str = str(e) if "execution reverted" in error_str or "VM Exception" in error_str: sell_blocked = "cannot" in error_str.lower() or "revert" in error_str.lower() if not result["buy_success"]: result["buy_error"] = error_str else: result["sell_error"] = error_str else: result["error"] = error_str # Determine honeypot verdict if result["buy_success"] and not result["sell_success"]: result["is_honeypot"] = True result["risk"] = "CRITICAL" elif result["buy_success"] and result["sell_success"] and result["details"].get("sell_tax_pct", 0) > 30: result["is_honeypot"] = False result["risk"] = "HIGH — extreme sell tax" elif result["buy_success"] and result["sell_success"]: result["is_honeypot"] = False result["risk"] = "LOW — both directions succeed" else: result["is_honeypot"] = None result["risk"] = "UNKNOWN — simulation error" return result def _check_blacklist_behavior(w3: Web3, token_address: str) -> dict: """Check if contract has blacklist functions by signature detection.""" code = w3.eth.get_code(Web3.to_checksum_address(token_address)).hex() blacklist_sigs = [ "9dc29fac", # blacklist(address) "10f41093", # setBlacklist(address,bool) "f9f92be4", # isBlacklisted(address) "4261f4d8", # addToBlacklist(address) "5d3a3f1b", # removeFromBlacklist(address) "ec571fc6", # setBlacklistStatus(address,bool) "e0021f43", # setExcludedFromFees(address,bool) "a733e570", # isExcludedFromFees(address) "b46300ec", # _isExcluded(address) "e42f6ea6", # listedAddresses(address) ] found = [] for sig in blacklist_sigs: if sig in code: found.append(f"0x{sig}") return {"blacklist_functions_detected": found, "has_blacklist": len(found) > 0} def _check_mint_function(w3: Web3, token_address: str) -> dict: """Check for mint function signature in contract bytecode.""" code = w3.eth.get_code(Web3.to_checksum_address(token_address)).hex() mint_sigs = ["40c10f19", "a0712d68", "9a1b1d3c", "d3fc0714", "f2d5d56b"] found = [] for sig in mint_sigs: if sig in code: found.append(f"0x{sig}") return {"mint_functions_detected": found, "has_mint": len(found) > 0} ``` ## Proxy / Upgradeability Detection Many scam tokens use proxy contracts (ERC-1967, TransparentUpgradeableProxy, BeaconProxy) to allow the deployer to swap the implementation contract after launch — effectively changing all token logic including balances, supply, and transfer restrictions. ```python def check_proxy_upgradeability(w3: Web3, token_address: str) -> dict: """ Detect if a token contract is behind a proxy pattern and read the implementation address. Checks ERC-1967, Transparent, UUPS, and Beacon patterns. """ address = Web3.to_checksum_address(token_address) result = { "is_proxy": False, "proxy_type": None, "implementation": None, "beacon": None, "admin": None, "can_upgrade": False, "details": {}, } # 1. ERC-1967 implementation slot (most common) try: impl_bytes = w3.eth.get_storage_at(address, ERC1967_IMPLEMENTATION_SLOT) impl_address = "0x" + impl_bytes[-20:].hex() if int(impl_bytes.hex(), 16) > 0: result["is_proxy"] = True result["proxy_type"] = "ERC-1967" result["implementation"] = Web3.to_checksum_address(impl_address) except Exception: pass # 2. Beacon proxy pattern if not result["is_proxy"]: try: beacon_bytes = w3.eth.get_storage_at(address, ERC1967_BEACON_SLOT) beacon_address = "0x" + beacon_bytes[-20:].hex() if int(beacon_bytes.hex(), 16) > 0: result["is_proxy"] = True result["proxy_type"] = "Beacon" result["beacon"] = Web3.to_checksum_address(beacon_address) # Read implementation from beacon (minimal ABI) beacon_abi = [{"inputs": [], "name": "implementation", "outputs": [{"type": "address"}], "stateMutability": "view", "type": "function"}] beacon_contract = w3.eth.contract(address=Web3.to_checksum_address(beacon_address), abi=beacon_abi) result["implementation"] = beacon_contract.functions.implementation().call() except Exception: pass # 3. Admin slot (who can upgrade?) if result["is_proxy"]: try: admin_bytes = w3.eth.get_storage_at(address, ADMIN_SLOT) admin_addr = "0x" + admin_bytes[-20:].hex() if int(admin_bytes.hex(), 16) > 0: result["admin"] = Web3.to_checksum_address(admin_addr) result["can_upgrade"] = True except Exception: pass # Check if implementation is a contract (i.e., has code) if result["implementation"]: impl_code = w3.eth.get_code(Web3.to_checksum_address(result["implementation"])) result["details"]["impl_has_code"] = len(impl_code) > 0 result["details"]["impl_code_size"] = len(impl_code) # Check for UUPS (Universal Upgradeable Proxy Standard) — proxy IS the implementation # UUPS has upgradeTo() in the implementation itself if result["implementation"]: try: impl_contract = w3.eth.contract( address=Web3.to_checksum_address(result["implementation"]), abi=[{"inputs": [{"type": "address"}], "name": "upgradeTo", "outputs": [], "stateMutability": "nonpayable", "type": "function"}], ) impl_contract.functions.upgradeTo(token_address).call({"from": Web3.to_checksum_address(result.get("admin") or "0x0000000000000000000000000000000000000000")}) result["details"]["appears_uups"] = True except Exception: result["details"]["appears_uups"] = False # 4. Check EIP-1167 minimal proxy (CREATE2 clones) code = w3.eth.get_code(address).hex() eip1167_prefix = "363d3d373d3d3d363d73" eip1167_suffix = "5af43d82803e903d91602b57fd5bf3" if eip1167_prefix in code and eip1167_suffix in code: result["is_proxy"] = True result["proxy_type"] = "EIP-1167 Minimal Proxy" # Extract target address from bytecode start = code.index(eip1167_prefix) + len(eip1167_prefix) impl_hex = code[start:start + 40] result["implementation"] = Web3.to_checksum_address("0x" + impl_hex) result["details"]["proxy_note"] = "EIP-1167 clone — immutable pointer, cannot upgrade" # 5. Compare proxy vs implementation events (if both available) if result["is_proxy"] and result["implementation"]: result["details"]["upgrade_risk"] = ( "CRITICAL: Admin can replace implementation at any time" if result["can_upgrade"] else "LOW: Proxy detected but no upgrade capability found" ) return result ``` ## Liquidity Lock Verification Checks if LP tokens are locked in a vesting/lock contract (Unicrypt, Team Finance, DXlock, or custom) and extracts lock parameters. ```python def verify_liquidity_lock( w3: Web3, lp_token_address: str, lock_contract_address: Optional[str] = None, chain: str = "ethereum", ) -> dict: """ Verify LP token lock status. If lock_contract_address is known, read from it directly. Otherwise scan common lock contracts on the chain. """ result = { "lp_locked": False, "lock_contract": None, "lock_end_date": None, "lock_amount": None, "lock_owner": None, "can_withdraw_early": None, "details": {}, } # Common lock contract ABI fragments lock_abi_fragments = [ # Unicrypt lock {"inputs": [{"type": "address"}], "name": "tokenLocks", "outputs": [{"type": "uint256"}, {"type": "uint256"}, {"type": "address"}, {"type": "bool"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"type": "uint256"}], "name": "lockedToken", "outputs": [{"type": "address"}, {"type": "address"}, {"type": "uint256"}, {"type": "uint256"}, {"type": "bool"}], "stateMutability": "view", "type": "function"}, {"inputs": [{"type": "address"}], "name": "getLocks", "outputs": [{"type": "uint256[]"}], "stateMutability": "view", "type": "function"}, # Team Finance (SaaS) generic {"inputs": [{"type": "address"}, {"type": "address"}], "name": "getLock", "outputs": [{"type": "uint256"}, {"type": "uint256"}, {"type": "address"}], "stateMutability": "view", "type": "function"}, ] # Try known lock contracts on Ethereum known_lock_contracts = { "ethereum": [ "0x17e00383A843A9922b8683F8F3D014cA41cC27b5", # Unicrypt v2 "0x663a5c229c09b049e36dCc11a9B0d4a8Eb9db214", # Unicrypt v1 "0xE2fE530C047f2d85298b07D9333C057D1Bf8E77b", # DXlock "0xa867C7C9e3438E7A5d6267a3f57b3910e00B2C25", # Team Finance ], "bsc": [ "0x663A5C229c09b049e36DcC11a9B0d4a8Eb9db214", # Unicrypt BSC "0xD7a3CECD1c95e40C2b0B4F9172DC9C490b6E2A3A", # PinkLock "0x71B7fF59207eD6a3cb7dE87b2d8a5C09e68a1C5f", # Team Finance BSC ], } targets = [lock_contract_address] if lock_contract_address else known_lock_contracts.get(chain, []) targets = [t for t in targets if t] for lock_addr in targets: try: lock_addr = Web3.to_checksum_address(lock_addr) lock_code = w3.eth.get_code(lock_addr) if lock_code.hex() == "0x": continue lock_contract = w3.eth.contract(address=lock_addr, abi=lock_abi_fragments[0]) # Try getLocks(address) — returns array of lock IDs try: lock_ids = lock_contract.functions.getLocks(lp_token_address).call() if lock_ids: for lock_id in lock_ids: try: lock_detail = lock_contract.functions.lockedToken(lock_id).call() result["lp_locked"] = True result["lock_contract"] = lock_addr result["lock_amount"] = lock_detail[2] result["lock_end_date"] = lock_detail[3] result["details"]["lock_id"] = lock_id # Determine if lock has expired current_time = int(time.time()) result["lock_expired"] = current_time > lock_detail[3] result["details"]["time_remaining_days"] = max(0, (lock_detail[3] - current_time) // 86400) break except Exception: continue except Exception: pass # If getLocks didn't work, try tokenLocks(address) if not result["lp_locked"]: try: lock_info = lock_contract.functions.tokenLocks(lp_token_address).call() result["lp_locked"] = True result["lock_contract"] = lock_addr result["lock_amount"] = lock_info[0] result["lock_end_date"] = lock_info[1] result["lock_owner"] = lock_info[2] except Exception: pass except Exception: continue if result["lp_locked"] and result["lock_end_date"]: result["details"]["formatted_end_date"] = time.strftime( "%Y-%m-%d %H:%M:%S UTC", time.gmtime(result["lock_end_date"]) ) return result ``` ## Sniper Bot Detection Identifies MEV/sniper bots that bought tokens in the first N blocks after the pool was created. These snipers often dump on retail buyers moments later. ```python def detect_sniper_activity( w3: Web3, token_address: str, pool_creation_block: int, lookback_blocks: int = 50, ) -> dict: """ Analyze the first N blocks after pool creation for sniper bot activity. Detects sandwich attacks, frontruns, and mass buy patterns. """ address = Web3.to_checksum_address(token_address) result = { "sniper_detected": False, "sniper_count": 0, "snipers": [], "first_trades": [], "suspect_patterns": [], "details": {}, } # 1. Get all Transfer events from pool creation to pool_creation + lookback_blocks transfer_event_sig = w3.keccak(text="Transfer(address,address,uint256)").hex() from_block = max(0, pool_creation_block - 5) # a few blocks before for LP add to_block = pool_creation_block + lookback_blocks try: logs = w3.eth.get_logs({ "address": address, "fromBlock": from_block, "toBlock": to_block, "topics": [transfer_event_sig], }) except Exception: # If get_logs range is too large, sample every 5th block logs = [] for b in range(from_block, to_block, 5): try: batch = w3.eth.get_logs({ "address": address, "fromBlock": b, "toBlock": min(b + 4, to_block), "topics": [transfer_event_sig], }) logs.extend(batch) except Exception: continue # 2. Parse events and identify buyers token = w3.eth.contract(address=address, abi=ERC20_ABI) buyer_tx_map = {} # buyer -> (block_number, tx_hash, is_first_buy) for log in logs: try: decoded = token.events.Transfer().process_log(log) if decoded["args"]["from"] == "0x0000000000000000000000000000000000000000": # Mint/buy event buyer = decoded["args"]["to"] tx_hash = log["transactionHash"].hex() block_num = log["blockNumber"] if buyer not in buyer_tx_map: buyer_tx_map[buyer] = { "address": buyer, "first_buy_block": block_num, "tx_hash": tx_hash, "amount": decoded["args"]["value"], "is_sniper_candidate": (block_num - pool_creation_block) <= 5, } else: buyer_tx_map[buyer]["amount"] += decoded["args"]["value"] except Exception: continue # 3. Identify sniper candidates weth_address = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2" for buyer, info in buyer_tx_map.items(): if info["is_sniper_candidate"]: # Check if this is a known MEV bot (contract with no code = EOA) code_size = len(w3.eth.get_code(Web3.to_checksum_address(buyer))) is_contract = code_size > 0 # Check if the buyer sold/dumped shortly after (look at outgoing transfers) sold_amount = 0 for log in logs: try: decoded = token.events.Transfer().process_log(log) if decoded["args"]["from"].lower() == buyer.lower() and decoded["args"]["to"] != "0x0000000000000000000000000000000000000000": sold_amount += decoded["args"]["value"] except Exception: continue info["is_contract"] = is_contract info["sold_amount"] = sold_amount info["is_dumper"] = sold_amount > info["amount"] * 0.5 # sold more than half if info["is_dumper"] or is_contract: result["snipers"].append(info) result["sniper_count"] += 1 # 4. Detect sandwich attack patterns # Look for same-block buy-sell pairs with different gas prices tx_blocks = {} for log in logs: block = log["blockNumber"] tx_hash = log["transactionHash"].hex() if block not in tx_blocks: tx_blocks[block] = set() tx_blocks[block].add(tx_hash) for block, txs in tx_blocks.items(): if len(txs) >= 3 and (block - pool_creation_block) <= 3: result["suspect_patterns"].append({ "type": "possible_sandwich", "block": block, "tx_count": len(txs), }) result["sniper_detected"] = result["sniper_count"] > 0 result["details"]["total_unique_buyers"] = len(buyer_tx_map) return result ``` ## Token Distribution Analysis (Gini Coefficient + Top Holders) Computes the Gini coefficient of token holder distribution — a high Gini (>0.9) indicates extreme concentration typical of scam tokens. ```python def analyze_holder_distribution(w3: Web3, token_address: str, top_n: int = 10) -> dict: """ Compute holder distribution metrics by scanning Transfer events. Returns Gini coefficient, top holder list, and concentration percentages. """ address = Web3.to_checksum_address(token_address) token = w3.eth.contract(address=address, abi=ERC20_ABI) result = { "total_holders": 0, "gini_coefficient": None, "top_holders": [], "top10_concentration_pct": 0.0, "top1_concentration_pct": 0.0, "supply_distribution": "UNKNOWN", "details": {}, } try: total_supply = token.functions.totalSupply().call() result["details"]["total_supply"] = total_supply except Exception: total_supply = 0 # Scan Transfer events to compute balances transfer_sig = w3.keccak(text="Transfer(address,address,uint256)").hex() balances = {} # Get latest block latest = w3.eth.block_number # Scan in chunks to avoid RPC limits chunk_size = 100000 from_block = max(0, latest - 500000) # last 500k blocks, adjust for recent tokens on new chains while from_block < latest: to_block = min(from_block + chunk_size - 1, latest) try: logs = w3.eth.get_logs({ "address": address, "fromBlock": from_block, "toBlock": to_block, "topics": [transfer_sig], }) for log_entry in logs: try: decoded = token.events.Transfer().process_log(log_entry) fr = decoded["args"]["from"].lower() to = decoded["args"]["to"].lower() val = decoded["args"]["value"] if fr == "0x0000000000000000000000000000000000000000" or fr == "0x0000000000000000000000000000000000000001": # Mint — skip, balance starts at receiving pass else: balances[fr] = balances.get(fr, 0) - val if balances[fr] <= 0: del balances[fr] balances[to] = balances.get(to, 0) + val except Exception: continue except Exception: # If chunk fails (too large), halve it chunk_size //= 2 if chunk_size < 1000: break continue from_block = to_block + 1 # Filter zero/negligible balances min_balance = total_supply / 1_000_000 if total_supply > 0 else 0 holders = {addr: bal for addr, bal in balances.items() if bal >= min_balance} if not holders: result["error"] = "Could not reconstruct holder balances from event logs" return result result["total_holders"] = len(holders) sorted_holders = sorted(holders.items(), key=lambda x: -x[1]) # Top holders result["top_holders"] = [ {"address": addr, "balance": bal, "pct": (bal / total_supply * 100) if total_supply > 0 else 0} for addr, bal in sorted_holders[:top_n] ] # Concentration percentages if total_supply > 0: top1_bal = sorted_holders[0][1] if sorted_holders else 0 result["top1_concentration_pct"] = top1_bal / total_supply * 100 top10_bal = sum(b for _, b in sorted_holders[:10]) result["top10_concentration_pct"] = top10_bal / total_supply * 100 # Gini coefficient values = sorted([b for _, b in sorted_holders if b > 0]) n = len(values) if n > 1: cumulative = [sum(values[:i+1]) for i in range(n)] # Gini = 2 * sum(i * y_i) / (n * sum(y_i)) - (n+1)/n numerator = sum((i + 1) * v for i, v in enumerate(values)) denominator = n * sum(values) gini = (2 * numerator / denominator) - (n + 1) / n result["gini_coefficient"] = round(gini, 4) # Distribution classification if result["top1_concentration_pct"] >= 90: result["supply_distribution"] = "CRITICAL — single holder controls >90%" elif result["top10_concentration_pct"] >= 90: result["supply_distribution"] = "HIGH — top 10 control >90%" elif result["top10_concentration_pct"] >= 50: result["supply_distribution"] = "MODERATE — top 10 control >50%" else: result["supply_distribution"] = "HEALTHY — distributed" return result ``` ## Wash Trading Detection (NFT) Detects circular trading patterns (A→B→C→A cycles) and same-wallet flipping in NFT sales data. ```python def detect_wash_trading( sales_data: List[Dict[str, Any]], min_cycle_length: int = 3, max_cycle_length: int = 10, ) -> dict: """ Detect wash trading patterns in NFT sales data. Input: list of dicts with {seller, buyer, token_id, price, timestamp, marketplace} Detects: circular trades, same-wallet flipping, and self-dealing. """ import networkx as nx from collections import defaultdict from itertools import combinations result = { "has_wash_trading": False, "circular_trades": [], "same_wallet_flips": [], "whale_wallets": [], "wash_volume": 0, "total_volume": sum(s.get("price", 0) for s in sales_data), "wash_volume_pct": 0.0, "details": {}, } if not sales_data: return result result["details"]["total_trades"] = len(sales_data) # 1. Build directed graph of trades G = nx.DiGraph() for sale in sales_data: seller = sale.get("seller", "").lower() buyer = sale.get("buyer", "").lower() price = sale.get("price", 0) token_id = sale.get("token_id", "") ts = sale.get("timestamp", 0) G.add_edge(seller, buyer, price=price, token_id=token_id, timestamp=ts) # 2. Detect cycles (circular trading) try: cycles = list(nx.simple_cycles(G)) except nx.NetworkXNoCycle: cycles = [] # Filter by cycle length filtered_cycles = [c for c in cycles if min_cycle_length <= len(c) <= max_cycle_length] result["circular_trades"] = [ {"cycle": [addr for addr in cycle], "length": len(cycle)} for cycle in filtered_cycles ] # 3. Detect same-wallet flipping (same wallet appearing as both buyer and seller across trades) wallet_activity = defaultdict(lambda: {"buys": 0, "sells": 0, "total_volume": 0, "token_ids": set()}) for sale in sales_data: seller = sale.get("seller", "").lower() buyer = sale.get("buyer", "").lower() price = sale.get("price", 0) token_id = sale.get("token_id", "") wallet_activity[seller]["sells"] += 1 wallet_activity[seller]["total_volume"] -= price # selling loses token, gains money wallet_activity[buyer]["buys"] += 1 wallet_activity[buyer]["total_volume"] += price if token_id: wallet_activity[seller]["token_ids"].discard(token_id) wallet_activity[buyer]["token_ids"].add(token_id) # Wallets with high buy-sell ratio across same tokens = wash traders for addr, activity in wallet_activity.items(): overlap = len(activity["token_ids"]) total_trades = activity["buys"] + activity["sells"] if total_trades >= 5 and overlap >= 3: result["whale_wallets"].append({ "address": addr, "buys": ac …(truncated)