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

# 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

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": ac

…(truncated)