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.
Instrucciones de origen · Vista previa de solo lectura
name
token-nft-scam-investigation
description
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.
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)
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
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.
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.
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.
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.
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.
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.
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": activity["buys"],
"sells": activity["sells"],
"overlapping_token_ids": overlap,
"suspected_wash_trader": overlap >= total_trades * 0.3,
})
# 4. Detect self-trading (wallet sells to itself via different addresses)
# Look for pairs of wallets that trade the same tokens back and forth
wallet_pairs = defaultdict(float)
for sale in sales_data:
seller = sale.get("seller", "").lower()
buyer = sale.get("buyer", "").lower()
price = sale.get("price", 0)
if seller < buyer:
pair = (seller, buyer)
else:
pair = (buyer, seller)
wallet_pairs[pair] += price
for (addr1, addr2), volume in wallet_pairs.items():
if volume > result["total_volume"] * 0.05: # more than 5% of total volume
result["details"]["suspicious_pair_trading"] = result["details"].get("suspicious_pair_trading", [])
result["details"]["suspicious_pair_trading"].append({
"wallet_a": addr1,
"wallet_b": addr2,
"volume": volume,
})
# 5. Compute wash volume
wash_volume = 0.0
for cycle in result["circular_trades"]:
for i in range(len(cycle)):
seller = cycle[i]
buyer = cycle[(i + 1) % len(cycle)]
for sale in sales_data:
if sale.get("seller", "").lower() == seller and sale.get("buyer", "").lower() == buyer:
wash_volume += sale.get("price", 0)
result["wash_volume"] = wash_volume
result["wash_volume_pct"] = (wash_volume / result["total_volume"] * 100) if result["total_volume"] > 0 else 0
result["has_wash_trading"] = len(result["circular_trades"]) > 0 or len(result["whale_wallets"]) > 2
return result
Deployer Funding Chain
Traces the deployer's initial funding transaction back through its chain to identify the ultimate source — an exchange deposit, another scam, a mixer, or a known malicious address.
def trace_deployer_funding(w3: Web3, deployer_address: str, max_hops: int = 5) -> dict:
"""
Trace the deployer's initial ETH funding back to its source.
Follows the first incoming ETH transaction at each hop.
"""
address = Web3.to_checksum_address(deployer_address)
result = {
"deployer": address,
"funding_chain": [],
"source_category": "UNKNOWN",
"source_exchange": None,
"source_suspicious": False,
"details": {},
}
# Known exchange deposit addresses (simplified list)
exchange_addresses = {
"0x3f5CE5FBFe3E9af3971dC833d26bA9b5C936f0bE": "Binance",
"0xdAC17F958D2ee523a2206206994597C13D831ec7": "Coinbase",
"0xBE0eB53F46cd790Cd13851d5EFf43D12404d33E8": "Binance 2",
"0x2faf487A4414Fe77e2327F0f0C2B1C1Ae3B6f5F": "Kraken",
"0x21a31Ee1afC51d94C2eFcCAa2092aD102858554": "OKX",
"0x4Baa4Ab5cBaa9C6CB2C7C6fE4D8a0F7F5B3F1E2": "KuCoin",
"0xf977814e90dA44bFA03b6295A0616a897441aceC": "Huobi",
"0x1Db3439a222C451ab00E0C2b3F2E0c7Ff5b5F5F5": "Gate.io",
"0x28C6c06298d514Db089934071355E5743bf21d60": "Binance 14",
"0x3Ed3B47Dd13EC9A98a44E6204A523a7669b1c6a5": "HTX",
}
# Known mixer/tumbler addresses
mixer_addresses = {
"0x8589427373D6D84E98730D7795D8f6f8731FDA16": "Tornado Cash",
"0x7221E6C5279B6681333FcaC7F7F8B5D0E1B0c5A1": "Tornado Cash",
"0x03893a7c7461F5C45C1D0e2E0D2a7B3A5F3B5F3B": "FixedFloat",
"0x1E2f4dF6b7A5c3E8C9A0b5D7F3E1C9A0b5D7F3E1": "Sinbad.io",
}
current = address
chain = []
for hop in range(max_hops):
try:
# Get first incoming ETH transaction to this address
# Using eth_getLogs for Transfer events won't work for ETH — we need tx data
# Simple approach: check first tx in block explorer or scan via eth_getBlockByNumber
nonce = w3.eth.get_transaction_count(current)
if nonce == 0:
# No outgoing txs — likely a fresh wallet
chain.append({"address": current, "type": "fresh_wallet", "note": "No outgoing transactions"})
if hop == 0:
result["source_category"] = "FRESH_WALLET"
break
# Find first tx where current was recipient
latest = w3.eth.block_number
found_tx = None
# Scan recent blocks for incoming ETH (limited scope — real impl needs archive node)
for block_num in range(latest, max(0, latest - 50000), -100):
try:
block = w3.eth.get_block(block_num, full_transactions=True)
for tx in block.transactions:
if isinstance(tx, dict) and tx.get("to", "").lower() == current.lower():
found_tx = tx
break
elif hasattr(tx, "to") and tx.to and tx.to.lower() == current.lower():
found_tx = tx
break
except Exception:
continue
if found_tx:
break
if not found_tx:
chain.append({"address": current, "type": "no_incoming_found", "note": "Could not find incoming ETH tx"})
result["source_category"] = "ARCHIVE_NODE_REQUIRED"
break
# Classify the sender
sender = (found_tx.get("from", "").lower() if isinstance(found_tx, dict)
else found_tx["from"].lower())
value = w3.from_wei(found_tx.get("value", 0) if isinstance(found_tx, dict) else found_tx["value"], "ether")
sender_lower = Web3.to_checksum_address(sender).lower() if sender else ""
# Check if sender is known
source_type = "wallet"
for exch_addr, exch_name in exchange_addresses.items():
if exch_addr.lower() == sender_lower:
source_type = f"exchange:{exch_name}"
result["source_exchange"] = exch_name
result["source_category"] = "EXCHANGE_DEPOSIT"
break
for mix_addr, mix_name in mixer_addresses.items():
if mix_addr.lower() == sender_lower:
source_type = f"mixer:{mix_name}"
result["source_suspicious"] = True
result["source_category"] = "MIXER"
break
chain.append({
"address": current,
"funded_by": sender,
"amount_eth": float(value),
"tx_hash": found_tx.get("hash", "").hex() if isinstance(found_tx, dict) else found_tx["hash"].hex(),
"source_type": source_type,
})
# Continue tracing if sender is not an exchange
if source_type in ("wallet",) or source_type.startswith("mixer"):
current = Web3.to_checksum_address(sender)
else:
break
except Exception as e:
chain.append({"address": current, "error": str(e)})
break
result["funding_chain"] = chain
return result
def check_deployer_scam_history(w3: Web3, deployer_address: str, api_key: str = ETHERSCAN_API_KEY) -> dict:
"""
Check if the deployer wallet has deployed other tokens — and if those
tokens have scam indicators.
"""
address = Web3.to_checksum_address(deployer_address)
result = {
"deployer": address,
"total_deployments": 0,
"previous_suspicious_tokens": [],
"is_serial_scammer": False,
"details": {},
}
# Query Etherscan for deployed contracts by this address
url = f"https://api.etherscan.io/api"
params = {
"module": "account",
"action": "txlist",
"address": address,
"startblock": 0,
"endblock": 99999999,
"sort": "asc",
"apikey": api_key,
}
try:
resp = requests.get(url, params=params, timeout=30)
data = resp.json()
if data.get("status") == "1" and data.get("result"):
deployments = []
for tx in data["result"]:
if tx.get("to", "") == "" and tx.get("contractAddress", ""):
# This transaction created a contract
deployments.append({
"contract": tx["contractAddress"],
"block": int(tx["blockNumber"]),
"tx_hash": tx["hash"],
"timestamp": int(tx.get("timeStamp", 0)),
})
result["total_deployments"] = len(deployments)
result["details"]["all_deployments"] = deployments[:20] # cap for output
# Check each deployment for scam indicators (basic)
for dep in deployments[:10]:
try:
info = check_basic_info(w3, dep["contract"])
if "ERROR" in str(info.get("name", "")):
continue
dep["name"] = info.get("name", "?")
dep["symbol"] = info.get("symbol", "?")
# Flag short-lived tokens
if dep.get("timestamp"):
age_days = (int(time.time()) - dep["timestamp"]) / 86400
dep["age_days"] = round(age_days, 1)
if age_days < 30:
dep["short_lived"] = True
except Exception:
continue
result["previous_suspicious_tokens"] = [d for d in deployments if d.get("short_lived")]
result["is_serial_scammer"] = len(result["previous_suspicious_tokens"]) >= 3
except Exception as e:
result["error"] = str(e)
return result
Approval / Allowance Abuse Detection
Checks if a token contract has abnormal approval patterns: unlimited approvals (max uint256), approvals to suspicious contracts, or backdoor approval functions.
def check_approval_abuse(w3: Web3, token_address: str, sample_holders: Optional[List[str]] = None) -> dict:
"""
Check for excessive or suspicious approval patterns in a token contract.
"""
address = Web3.to_checksum_address(token_address)
token = w3.eth.contract(address=address, abi=ERC20_ABI)
result = {
"has_unlimited_approvals": False,
"suspicious_spenders": [],
"approval_event_analysis": {},
"details": {},
}
max_approval = 2**256 - 1 # unlimited approval sentinel
# 1. Check allowance for known risky spenders
suspicious_spenders = [
"0x0000000000000000000000000000000000000001", # null address
"0x000000000000000000000000000000000000dead", # burn address
]
# 2. Analyze Approval events to find patterns
approval_sig = w3.keccak(text="Approval(address,address,uint256)").hex()
latest = w3.eth.block_number
try:
logs = w3.eth.get_logs({
"address": address,
"fromBlock": max(0, latest - 200000),
"toBlock": latest,
"topics": [approval_sig],
})
approval_counts = {}
unlimited_approvals = 0
total_approvals = 0
for log in logs:
try:
decoded = token.events.Approval().process_log(log)
owner = decoded["args"]["owner"].lower()
spender = decoded["args"]["spender"].lower()
value = decoded["args"]["value"]
total_approvals += 1
# Track approval count per spender
approval_counts[spender] = approval_counts.get(spender, 0) + 1
# Check unlimited approvals
if value == max_approval:
unlimited_approvals += 1
# Check suspicious spenders
if spender in suspicious_spenders:
result["suspicious_spenders"].append({
"owner": owner,
"spender": spender,
"value": value,
"is_unlimited": value == max_approval,
})
except Exception:
continue
result["approval_event_analysis"] = {
"total_approval_events": total_approvals,
"unlimited_approvals": unlimited_approvals,
"unique_spenders": len(approval_counts),
"most_approved_spender": max(approval_counts, key=approval_counts.get) if approval_counts else None,
}
result["has_unlimited_approvals"] = unlimited_approvals > 0
except Exception as e:
result["details"]["approval_scan_error"] = str(e)
# 3. Check for approval-related function signatures in bytecode
code = w3.eth.get_code(address).hex()
approval_sigs = [
"095ea7b3", # approve(address,uint256)
"a9059cbb", # transfer(address,uint256)
"23b872dd", # transferFrom(address,address,uint256)
"dd62ed3e", # allowance(address,address)
]
for sig in approval_sigs:
result["details"][f"has_{sig[:4]}"] = sig in code
return result
Step 3: NFT Wash Trading Detection
For NFT collections, detect market manipulation:
Same-wallet cycling — NFT moving between same wallets repeatedly
Circular trading — wallet A sells to B, B sells to C, C sells to A
Price laddering — artificially inflating floor prices through self-deals
Volume farming — generating fake volume to attract organic buyers
Cross-marketplace arbitrage manipulation — wash trading across platforms
Step 4: Social and On-Chain Correlation
Cross-reference on-chain data with off-chain signals:
Team doxxing status — is the team known or anonymous?
Social media hygiene — are there red flags on Twitter, Discord, Telegram?
Deployer wallet history — has this wallet deployed other scam tokens?
Funding source — where did the deployer get initial funds? Connected to other scams?
Step 5: Scam Proceeds Tracing
Track where victim funds went:
Liquidity removal — when LP was pulled and where the funds went
DEX dumping — sell transactions by the deployer wallet
Bridge/CEX deposit — funds moved to a bridge or centralized exchange for cash-out
Cross-chain transfer — funds moved to another chain to obscure trail
Case Studies
Case Study 1: Squid Game Token (SQUID) — Honeypot + Rug Pull
Proxy contract (ERC-1967 pattern) — implementation could be swapped at any time
pause() and unpause() functions present in bytecode — transfers could be frozen
Sell function was disabled via the proxy implementation after enough buy pressure
Top 10 holders controlled >99% of supply at launch
Deployer funded from Tornado Cash
Website created 30 days before launch, no team info
LP was locked for only 30 days with an early-withdraw clause
What happened: The token skyrocketed 230,000% in days. When holders tried to sell, the proxy implementation was swapped to one that blocked all sells. The deployer extracted ~$3.38M in BNB from the liquidity pool and bridged to Ethereum.
Detection code applied: simulate_honeypot() would have detected the sell failure. check_proxy_upgradeability() would have shown the non-upgradeable facade. trace_deployer_funding() would have identified the Tornado Cash connection.
Ownership not renounced — deployer retained owner() control
Withdraw function not removed from the contract — allowed draining of mint funds
No lock on the dev wallet address
Deployer wallet had deployed 3 other unknown NFT projects that were abandoned
No multi-sig — deployer was a single EOA
"Fight Arena" P2E game was promised but no development wallet activity
What happened: The anonymous developer ("Feisty Doge" / "Evil Ape") minted all 10,000 NFTs, emptied the mint contract of 798 ETH via the withdraw() function, deleted the website and Discord, and disappeared. The NFTs became worthless.
Detection code applied: check_all_ownership_interfaces() would have shown non-renounced ownership. check_deployer_scam_history() would have flagged the 3 previous abandoned projects. The contract's withdraw() function signature (0x2e1a7d4d) would have been detected.
Case Study 3: Frosties NFT — Wash Trading + Rug Pull
Active wash trading on secondary markets — 40% of volume was circular trades between known wallets
Two wallet addresses controlled 6 of the top 10 traded NFTs
Deployer funded from a separate rug pull (crypto panda NFT)
Multi-sig was a 2/3 with two addresses linked to other rug pulls
Discord had 30,000+ members and heavily moderated criticism
What happened: After the mint sold out (~$1.1M), the team transferred all funds to a separate wallet, bridged ETH to Bitcoin via a swap service, and shut down operations. The founders ("Frostie" and "N0x") were later arrested by the DOJ in 2023.
Detection code applied: detect_wash_trading() would have caught the 40% wash volume. trace_deployer_funding() would have connected the deployer to the earlier crypto panda rug. The multi-sig addresses would have shown the connection when checked by check_deployer_scam_history().
Case Study 4: AnubisDAO (Anubis) — Honeypot + LP Removal
OHM fork with disguised tokenomics — the peg mechanism was non-functional
LP tokens were never locked — deployer held both sides of the LP
Sushiswap LP was added with 13,576 ETH from a multi-sig
Multi-sig signers were anonymous with no public identity
Website was copied from another OHM fork with minor edits
Stake contract contained a "setRewardDistribution" function that could drain staking rewards
What happened: Within hours of the LP being added, someone (perpetrator or inside job) removed all 13,576 ETH from the LP pool. The token became worthless. The stolen ETH was split across several addresses and ultimately deposited into multiple exchanges.
Detection code applied: verify_liquidity_lock() would have shown ZERO LP locks — the LP was never locked. The lack of lock should have been a CRITICAL red flag. The multi-sig analysis would have shown the owner controlled both LP tokens.
Off-Chain Investigation — Cross-reference deployer history, social signals, and funding sources
Reporting — Produce structured investigation report with evidence and confidence scores
Verification Loop — Cross-check each finding with at least one additional data source before finalizing
Money Extraction — Package findings into client deliverables or bounty submissions (see Money section)
Verification
Honeypot check verified by simulating both buy AND sell transactions (with actual gas estimation)
LP lock status confirmed by reading the lock contract directly, not just a third-party claim
Wash trading detection verified by checking actual wallet ownership on-chain, not just marketplace labels
Deployer connections confirmed by tracing initial funding transaction back to the source
Every claim in the report supported by a verifiable on-chain transaction or external link
Top holder analysis verified by cross-referencing Transfer events from two different block ranges
Proxy detection verified by reading storage slots directly (not relying on event logs)
Sniper detection verified by checking actual gas prices and block positions at launch
Funding chain verified by checking at least two hops before classifying the source
RPC & API Endpoint Management
Token investigations span multiple chains. A scam token on Ethereum often has a matching token on BSC
or Polygon via cross-chain deployment. You need reliable RPC access on all relevant chains.
Per-Chain Rate Limits
Chain
Explorer API
Free Limit
Key Env Var
Ethereum
Etherscan
5 calls/sec
ETHERSCAN_KEY
BSC
BscScan
5 calls/sec
BSCSCAN_KEY
Polygon
Polygonscan
5 calls/sec
POLYGONSCAN_KEY
Arbitrum
Arbiscan
10 calls/sec
ARBISCAN_KEY
Optimism
Optimistic Etherscan
5 calls/sec
OPSCAN_KEY
Avalanche
Snowtrace
5 calls/sec
SNOWTRACE_KEY
Rule: batch independent checks (holdership, LP lock, ownership, honeypot) into a single function
that respects the slowest rate limit — queue at 4 calls/sec to be safe across all chains.
Contract Metadata via RPC (No Explorer Key)
from web3 import Web3
# Public RPC endpoints — no key needed
PUBLIC_RPC = {
"ethereum": "https://eth.llamarpc.com",
"bsc": "https://bsc-dataseed.binance.org",
"polygon": "https://polygon-rpc.com",
}
def check_contract_metadata(chain: str, address: str) -> dict:
"""Query contract metadata directly from RPC — useful when explorer API is rate-limited."""
w3 = Web3(Web3.HTTPProvider(PUBLIC_RPC[chain]))
checksum = w3.to_checksum_address(address)
code = w3.eth.get_code(checksum)
if code == b"": # EOA or non-existent — no contract
return {"has_contract": False}
# Get bytecode hash for heuristic clone detection
import hashlib
code_hash = hashlib.sha256(code).hexdigest()
return {"has_contract": True, "code_hash": code_hash, "byte_size": len(code)}
Cross-Chain Token Address Discovery
When you find a scam token on one chain, the same deployer may have identical tokens on other chains.
Use the deployer address to cross-reference:
def find_cross_chain_tokens(deployer: str, chains: list[str] = None) -> dict:
"""Search known deployers across chains for same-token patterns."""
results = {}
for chain in (chains or ["ethereum", "bsc", "polygon"]):
rpc = PUBLIC_RPC.get(chain)
if not rpc:
continue
w3 = Web3(Web3.HTTPProvider(rpc))
# Get deployer's nonce at deployment block to compute deployed addresses
# (Each deployment increments nonce, address = RLP(deployer, nonce))
# Simplified: check if the deployer has any token contract on this chain
tx_count = w3.eth.get_transaction_count(w3.to_checksum_address(deployer))
results[chain] = {"chain_active": True, "deployer_tx_count": tx_count}
return results
Transaction Trace Decision Tree
Not every suspicious transaction is a scam. Before committing to a full investigation, classify the
transaction pattern:
Signal
Likely Type
Recommended Action
Deployer → addLiquidity → renounceOwnership
Classic rug pull
Full investigation (deployer history, LP lock, tx analysis)
Deployer → multiSend (100+ addresses)
Airdrop / distribution
Check tokenomics; likely legitimate if paired with socials
Flash loan → swap → deposit to contract
MEV or attack
Use defi-incident-analysis skill instead
Small buys from 10+ fresh wallets in same block
Wash trading
Check for identical gas prices / same funding source
Developer mints 100% supply → lock to vesting
Cautionary but might be legit
Verify the lock contract allows no early-withdraw
Contract has _beforeTokenTransfer hook
Potential honeypot
Simulate sell with Tenderly before buying
When to Stop
Three hard exit criteria for token investigations:
The token is confirmed as a known scam variant (same bytecode hash as a registry of known scam
contracts) → document and report; no need to reconstruct every detail
The deployer address has been blacklisted on 3+ chain explorers → report is complete; the
scammer is known
The total scam value is under $1,000 → one-page summary, no deep trace
Money Section
Sell token/NFT scam investigation services through the following channels:
Court testimony + expert declaration for fraud cases (varies by case)
Client Acquisition
Rug pull victims — Monitor Discord servers and Telegram groups of recent rug pulls. Offer a "post-mortem analysis" for $200 that traces where their money went. Victims are highly motivated and can share the report with lawyers.
Pre-purchase due diligence — Offer a "token safety audit" for $100-300 before investors buy into a new launch. Market in crypto alpha groups, Discord launch servers, and Telegram trading communities.
NFT buyers — NFT collectors will pay $50-100 to check if a collection has wash trading before buying. Market in NFT Discord communities and Twitter/X.
Insurance claims — Projects that claim to have been rugged need proof for insurance. Offer "scam verification reports" for $500-1,000.
Law firms — Lawsuits against crypto projects need expert declarations. Charge $5,000-10,000 for a court-ready report with chain-of-custody documentation.
Bounty platforms — Some projects and forums offer bounties for identifying scam tokens (e.g., Immunefi for yield farms, community bounties for Telegram groups).
Marketing Angle
"I traced the deployer's wallet to a known serial scammer"
"The token passed all basic checks but failed the proxy upgradeability test"
"80% of the trading volume was wash trading between 3 wallets"
"The LP lock contract has an early-withdraw function only the deployer knows about"
"Your tax records: here is the on-chain proof of the rug pull for the IRS"
Workflow for Client Delivery
Client provides token/NFT contract address and a brief (what to investigate)
Run run_scam_check() to produce initial findings
If positive (scam detected), deep-dive with deployer tracing and fund flow analysis
Produce a PDF report with: executive summary, methodology, findings per category, screenshots of key transactions, and a chain-of-custody document
Deliver with a live walkthrough (Zoom, Google Meet) for Premium tier clients
Follow up after 30 days — if the project rugged, the client is motivated to buy the deep trace
Practices from Top Investigators
The ZachXBT Methodology
ZachXBT has investigated hundreds of token and NFT scams, identifying the same serial scammers
operating across multiple projects. His key insight: scammers reuse everything — wallet addresses,
funding sources, website templates, Telegram handles. The first step in any token investigation
is checking if the deployer or any related wallet has been seen before.
Identifying if the deployer has been tagged before
Blockchair
BTC privacy analysis for crypto ransom tokens
BTC-based scam tracing
OMNIA
Mempool monitoring: catch deployer's next tx before it confirms
Counter-party surveillance
MetaSuites
Legacy labeling for historical address lookups
Old addresses (pre-2022) may only have labels here
Pattern Recognition Across Multiple Scams
Serial scammers exhibit consistent patterns. When investigating a new token/NFT, cross-reference:
PATTERN_SIGNATURES = {
"same_deployer": {
"check": "Deployer address used in >1 token launch",
"severity": "CRITICAL — confirmed serial deployer",
"action": "Flag all tokens from this deployer as high-risk immediately",
},
"same_funding_source": {
"check": "Deployer funded from an address linked to a known scam",
"severity": "HIGH — strong serial scammer signal",
"action": "Trace the funding address for additional scam tokens",
},
"same_website_domain": {
"check": "Token website = registered by same email / same IP as known scam",
"severity": "CRITICAL — direct identity link",
"action": "WHOIS lookup the domain; check registrar history",
},
"same_telegram_discord": {
"check": "Same social handles used in previous scam projects",
"severity": "HIGH — the scammers are running it back",
"action": "Archive the prior scam thread for evidence chain",
},
"same_twitter_handle": {
"check": "Twitter account reused across projects",
"severity": "HIGH — identity link, handle may be deleted",
"action": "Capture via archive.org / screenshot immediately",
},
"same_bytecode_hash": {
"check": "Contract bytecode matches a known scam contract exactly",
"severity": "HIGH — copy-paste scam",
"action": "No need to analyze the code; reference the prior analysis",
},
"same_liquidity_pattern": {
"check": "LP added then removed in identical timing to previous scam",
"severity": "MEDIUM — behavioral match",
"action": "Correlate with other signatures before confirming",
},
}
Social Engineering Deconstruction
ZachXBT documents how scammers manipulate victims. Understanding these patterns helps you
predict where to find evidence:
Common scammer playbooks in token/NFT fraud:
The Fake Influencer: Impersonation accounts DM victims with "exclusive token" → fake website → malicious approve() → wallet drained
Evidence trail: The impersonated account's followers; the real handle's report; common scam contract deployer
The Discord "Admin": Hacked Discord server → admin DMs victims with "mint now" link → website wallet drain
The NFT "Rug": Generative art project → hyped mint → floor starts at 0.01 ETH → team stops responding after mint sell
Evidence trail: Team wallet minting multiple NFTs to self; immediate listings at above floor; Discord/Twitter deletion
def classify_scammer_playbook(deployer_tx_history: list[dict]) -> list[str]:
"""Classify which playbook(s) a scammer is using based on deployer behavior."""
playbooks = []
tx_types = [tx.get("functionName", "") for tx in deployer_tx_history]
tx_count = len(tx_types)
# Check for liquidity trap pattern
approve_txs = [t for t in tx_types if "approve" in t.lower()]
lp_txs = [t for t in tx_types if "addLiquidity" in t.lower() or "add_liquidity" in t.lower()]
if len(lp_txs) >= 2 and len(approve_txs) > len(lp_txs):
playbooks.append("liquidity_trap")
# Check for multi-token deployer (serial deployer)
from collections import Counter
contract_creations = [t for t in tx_types if "create" in t.lower()]
if len(contract_creations) >= 3:
playbooks.append("serial_deployer")
# Check for approve-drain pattern (phishing)
if tx_count < 20 and len(approve_txs) >= 3:
playbooks.append("approve_drain_phishing")
return playbooks or ["unknown"]
Funding Chain Tracing for Token Scams
Apply the funding chain method specifically to token deployers:
Find deployer address — Get from address of token creation transaction
Trace deployer's first funding — Where did the deployer get ETH for gas?
Cross-check funding source against known scam databases (check each prior hop)
If the funder is a known scammer, all tokens from this deployer are high-risk
If the funder is a CEX — note the exchange; this is a potential KYC link
The Investigation-to-Bounty Pipeline
Modeled on ZachXBT's proven approach:
Beta token check (free, public) — Post a thread on Twitter/X with your initial analysis
If scam confirmed — Publish the full wallet cluster thread with evidence
Tag relevant parties — The project's accounts, the chain's security team, exchange accounts
Bounty collection — Some projects and DAOs offer bounties for scam identification
Private client referrals — Each investigation builds reputation for paid work
Anti-Rationalization Table
Anti-Rationalization Table
Rationalization
Reality
"The token has a verified contract, it's legit"
Verification only proves the source matches bytecode, not that the logic is honest. Many verified contracts are honeypots with legitimate-looking source that hides malicious logic in proxy implementations.
"The team is doxxed, they wouldn't rug"
Doxxed teams have rugged too. Credentials don't guarantee integrity. Multiple projects with KYCed teams have rugged (e.g., some Certik KYC projects).
"LP is locked so it's safe"
Locked LP prevents liquidity removal but doesn't prevent minting, proxy upgrades, or blacklist abuse. Also, locks can have early-withdrawal functions or be owned by the deployer.
"Low market cap means less risk"
Low cap tokens are the most common rug pull targets. Market cap has no correlation with safety — many rug pulls start with very low caps to attract initial buyers.
"The contract is renounced, so it can't be changed"
Renouncing only removes the owner() role. If the contract uses AccessControl, the DEFAULT_ADMIN_ROLE can still mint. Proxy contracts have a separate admin that can upgrade the implementation.
"High trading volume means it's legit"
Volume can be entirely wash-traded. Wash trading detection in this skill shows how 40%+ of volume can be circular trades between controlled wallets, creating a false appearance of organic demand.
"The team passed KYC with a reputable firm"
KYC verification is only as good as the KYC provider. Some KYC'd teams have still rugged (the KYC firm retains liability but doesn't prevent the scam). Also, KYC credentials can be stolen or fake.
"Liquidity is in multiple DEXes, so removing it all is impossible"
Deployers can coordinate removal across DEXes or use flash loans to drain all pools simultaneously. Multiple pools don't mean any single one is safe.
"The code was audited by a reputable firm"
Audits only check what the auditor was asked to check. Scam tokens often get audited with the malicious code excluded from scope. Audits don't guarantee safety.
"It's a well-known project with a large community"
Community size is easily faked with bots. Many rug pulls had 30,000+ Discord members (90%+ bots). Large communities create false social proof.
