# Onchain Transaction Forensics > Use when trace and analyze blockchain transactions to investigate illicit fund flows, identify wallet clusters, and map transaction graphs across multiple blockchains. Use when investigating stolen funds, following money trails on-chain, analyzing suspicious addresses, or tracing cross-chain transactions. - Skill: `oyi77/onchain-transaction-forensics` (Agent Skill) - Install (CLI): `npx skillmds add oyi77/onchain-transaction-forensics` - Raw SKILL.md: https://api.skillmd.com/api/skills/oyi77/onchain-transaction-forensics/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: Apache-2.0 - Author: oyi77 (https://skillmd.com/u/oyi77) - Updated: 2026-09-08 - Page: https://skillmd.com/skills/oyi77/onchain-transaction-forensics --- # On-Chain Transaction Forensics ## Overview Blockchain transaction forensics involves tracing and analyzing cryptocurrency transactions to investigate illicit fund flows, identify wallet clusters, and map transaction graphs across multiple chains. Unlike traditional financial forensics, blockchain provides a public, immutable ledger where every transaction is permanently recorded. Every hop a stolen fund takes — from initial theft through mixers, bridges, DEX swaps, and exchange deposits — leaves indelible evidence. This skill covers collecting transaction data via block explorer APIs and RPC nodes, building directed transaction flow graphs with NetworkX, applying clustering heuristics (peel chain detection, consolidation patterns, CoJoin clustering for UTXO chains, address reuse), identifying mixing services and cross-chain bridges by scanning event logs and known contract registries, and producing investigator-ready trace reports with Sankey flow visualization. The techniques here apply to EVM chains (Ethereum, BNB Chain, Polygon, Arbitrum, Optimism, Avalanche), Bitcoin-family UTXO chains, Solana, and TRON. ## When to Use **Trigger phrases:** - "onchain transaction forensics" - "Trace stolen funds on the blockchain" - "Follow the money on-chain" - "Analyze suspicious wallet transactions" - "Where did the stolen crypto go" - "Fund flow analysis" - "Blockchain investigation" - When investigating theft, rug pull, or hack incidents that require tracing stolen cryptocurrency - When building evidence chains for law enforcement or bounty submissions (Arkham, Lazarus Bounty) - When analyzing suspicious wallet activity for compliance or risk assessment - When reconstructing the flow of funds through mixers, bridges, or exchanges - When preparing transaction trace reports as a paid forensic service - When mapping the flow of illicit funds from a known attacker address - When performing due diligence on a large transaction or wallet before a business deal ## When NOT to Use - You need real-time monitoring of transactions (use a blockchain monitoring service or webhook provider) - Task requires recovering private keys or seed phrases (use wallet recovery tools) - You need to analyze smart contract logic, not transaction flow (use smart contract analysis skills) - The blockchain is privacy-focused (Monero, Zcash) where on-chain analysis is limited to public data - You don't have access to a reliable RPC node or block explorer API - The investigation requires off-chain data (social media, KYC records, exchange logs) that you cannot correlate - You are analyzing a DeFi protocol for smart contract vulnerabilities rather than following fund movements ## Prerequisites - Python 3.8+ with web3.py, requests, pandas, networkx, numpy, plotly - Access to at least one Ethereum RPC node (Infura, Alchemy, or local node) - Etherscan API key (or equivalent block explorer API key for target chains) - Basic understanding of blockchain transaction structure (inputs, outputs, logs, events) - Graph visualization tools (Graphviz `dot` for DOT rendering, or plotly for HTML output) - For UTXO chains: understanding of inputs/outputs, addresses vs public keys, change addresses - At least 2 GB RAM for processing large transaction datasets (100k+ txs) - Basic knowledge of DeFi protocols, bridge contracts, mixer mechanics ## Multi-Chain Block Explorer Reference | Chain | Block Explorer | API Base URL | API Key Required | Notes | |---|---|---|---|---| | Ethereum | Etherscan | `https://api.etherscan.io/api` | Yes | Most comprehensive EVM explorer | | Ethereum (batch) | Etherscan V2 | `https://api.etherscan.io/v2/api` | Yes | Supports `chainid` param | | BNB Chain | BscScan | `https://api.bscscan.com/api` | Yes | Same API shape as Etherscan | | Polygon | Polygonscan | `https://api.polygonscan.com/api` | Yes | Same API shape as Etherscan | | Arbitrum | Arbiscan | `https://api.arbiscan.io/api` | Yes | Same API shape as Etherscan | | Optimism | Optimistic Etherscan | `https://api-optimistic.etherscan.io/api` | Yes | Same API shape as Etherscan | | Avalanche C-Chain | Snowtrace | `https://api.snowtrace.io/api` | Yes | Same API shape as Etherscan | | Base | Basescan | `https://api.basescan.org/api` | Yes | Same API shape as Etherscan | | TRON | Trongrid | `https://api.trongrid.io` | Yes (optional) | REST API, different structure | | Solana | Solscan | `https://api.solscan.io` | Yes | Different API from EVM explorers | | Bitcoin | Blockchair | `https://api.blockchair.com/bitcoin` | Yes (free tier) | UTXO-specific endpoints | | Bitcoin | mempool.space | `https://mempool.space/api` | No | Public, rate-limited | | Bitcoin | Blockchain.com | `https://blockchain.info` | No | Legacy API, rate-limited | | Multi-chain | Covalent | `https://api.covalenthq.com/v1` | Yes | Unified API across 100+ chains | | Multi-chain | BitQuery | `https://graphql.bitquery.io` | Yes | GraphQL, supports many chains | ### Etherscan-like API Wrapper The following chains use an identical API schema to Etherscan. A single helper class handles all of them: ```python import requests from typing import Optional EXPLORERS = { "ethereum": "https://api.etherscan.io/api", "bsc": "https://api.bscscan.com/api", "polygon": "https://api.polygonscan.com/api", "arbitrum": "https://api.arbiscan.io/api", "optimism": "https://api-optimistic.etherscan.io/api", "avalanche": "https://api.snowtrace.io/api", "base": "https://api.basescan.org/api", } class ExplorerClient: """Unified client for Etherscan-family block explorers.""" def __init__(self, api_key: str, chain: str = "ethereum"): self.api_key = api_key base = EXPLORERS.get(chain) if not base: raise ValueError(f"Unsupported chain: {chain}") self.base = base def _call(self, params: dict) -> list: params["apikey"] = self.api_key resp = requests.get(self.base, params=params, timeout=30) data = resp.json() if data.get("status") != "1": return [] return data.get("result", []) def normal_txs(self, address: str) -> list[dict]: return self._call({ "module": "account", "action": "txlist", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc" }) def internal_txs(self, address: str) -> list[dict]: return self._call({ "module": "account", "action": "txlistinternal", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc" }) def erc20_transfers(self, address: str) -> list[dict]: return self._call({ "module": "account", "action": "tokentx", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc" }) def erc721_transfers(self, address: str) -> list[dict]: return self._call({ "module": "account", "action": "tokennfttx", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc" }) def get_logs(self, address: str, from_block: int = 0, to_block: int = 99999999, topic0: Optional[str] = None) -> list[dict]: params = { "module": "logs", "action": "getLogs", "address": address, "fromBlock": from_block, "toBlock": to_block, } if topic0: params["topic0"] = topic0 return self._call(params) ``` ## Core Workflow: Data Collection ### Etherscan V2 Batch API The Etherscan V2 API supports multi-chain queries via a single endpoint by passing `chainid`. This reduces complexity when tracing funds across chains: ```python def get_txs_v2(address: str, api_key: str, chain_id: int = 1) -> list[dict]: """Fetch normal transactions via Etherscan V2 batch API.""" url = f"https://api.etherscan.io/v2/api" params = { "chainid": chain_id, "module": "account", "action": "txlist", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc", "apikey": api_key } resp = requests.get(url, params=params, timeout=30) return resp.json().get("result", []) def get_internal_txs_v2(address: str, api_key: str, chain_id: int = 1) -> list[dict]: """Fetch internal transactions via Etherscan V2 batch API.""" url = "https://api.etherscan.io/v2/api" params = { "chainid": chain_id, "module": "account", "action": "txlistinternal", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc", "apikey": api_key } resp = requests.get(url, params=params, timeout=30) return resp.json().get("result", []) def get_erc20_transfers_v2(address: str, api_key: str, chain_id: int = 1) -> list[dict]: """Fetch ERC-20 token transfers via Etherscan V2 batch API.""" url = "https://api.etherscan.io/v2/api" params = { "chainid": chain_id, "module": "account", "action": "tokentx", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc", "apikey": api_key } resp = requests.get(url, params=params, timeout=30) return resp.json().get("result", []) def get_token_balances_v2(address: str, api_key: str, chain_id: int = 1) -> list[dict]: """Fetch ERC-20 token balances via Etherscan V2.""" url = "https://api.etherscan.io/v2/api" params = { "chainid": chain_id, "module": "account", "action": "tokenlist", "address": address, "apikey": api_key } resp = requests.get(url, params=params, timeout=30) return resp.json().get("result", []) ``` **Chain IDs reference:** | Chain | chainid | |---|---| | Ethereum Mainnet | 1 | | BNB Chain | 56 | | Polygon | 137 | | Arbitrum One | 42161 | | Optimism | 10 | | Avalanche C-Chain | 43114 | | Base | 8453 | | Linea | 59144 | | Scroll | 534352 | ### Bitcoin Transaction Fetcher (mempool.space) ```python def get_btc_address_txs(address: str) -> list[dict]: """Fetch Bitcoin transaction history via mempool.space (no API key needed).""" url = f"https://mempool.space/api/address/{address}/txs" resp = requests.get(url, timeout=30) return resp.json() def get_btc_tx_details(txid: str) -> dict: """Fetch full Bitcoin transaction details.""" url = f"https://mempool.space/api/tx/{txid}" resp = requests.get(url, timeout=30) return resp.json() ``` ### TRON Transaction Fetcher ```python def get_tron_txs(address: str, api_key: str = "") -> list[dict]: """Fetch TRON transactions via Trongrid.""" headers = {"TRON-PRO-API-KEY": api_key} if api_key else {} url = f"https://api.trongrid.io/v1/accounts/{address}/transactions" resp = requests.get(url, headers=headers, timeout=30) return resp.json().get("data", []) ``` ## Step-by-Step Methodology ### Step 1: Collect All Transactions for Target Addresses Start with the primary address from the incident — the theft wallet, the hacker's known address from a Rug Pull, or the suspicious deposit address. ```python def collect_all_txs(address: str, api_key: str, chain_id: int = 1) -> dict: """Collect normal, internal, and ERC-20 tx history for one address.""" return { "normal": get_txs_v2(address, api_key, chain_id), "internal": get_internal_txs_v2(address, api_key, chain_id), "erc20": get_erc20_transfers_v2(address, api_key, chain_id), } ``` - **Normal TXs**: Standard ETH/BNB/MATIC transfers where `from` and `to` are directly visible - **Internal TXs**: Transfers triggered by contract CALL opcodes — these are NOT visible in normal tx lists but move ETH. Critical for tracing through DEX swaps, DeFi interactions, and bridge deposits - **ERC-20 transfers**: Token movements (`transfer` events). The most common value transfer mechanism in DeFi hacks - **ERC-721 / ERC-1155**: NFT transfers. Relevant for NFT theft and phishing investigations **Pagination**: Etherscan API returns up to 10,000 rows per call. For addresses with more transactions, pass the `page` parameter: ```python def collect_all_txs_paginated(address: str, api_key: str, chain: str = "ethereum", max_pages: int = 10) -> list[dict]: """Paginate through large transaction histories.""" client = ExplorerClient(api_key, chain) all_txs = [] for page in range(1, max_pages + 1): params = { "module": "account", "action": "txlist", "address": address, "startblock": 0, "endblock": 99999999, "sort": "asc", "page": page, "offset": 10000 } txs = client._call(params) if not txs: break all_txs.extend(txs) if len(txs) < 10000: break return all_txs ``` ### Step 2: Parse and Normalize Transaction Data Parse the raw API responses into a uniform format suitable for graph construction: ```python import pandas as pd def normalize_txs(raw_txs: list[dict], tx_type: str = "normal") -> pd.DataFrame: """Convert raw Etherscan API results to a normalized DataFrame.""" rows = [] for tx in raw_txs: row = { "hash": tx.get("hash"), "block": int(tx.get("blockNumber", 0)), "timestamp": int(tx.get("timeStamp", 0)), "from": tx.get("from", "").lower(), "to": tx.get("to", "").lower(), "value_wei": int(tx.get("value", 0)), "value_eth": int(tx.get("value", 0)) / 1e18, "gas": int(tx.get("gas", 0)), "gas_price": int(tx.get("gasPrice", 0)), "input": tx.get("input", ""), "is_error": tx.get("isError", "0") == "1", "tx_type": tx_type, } if tx_type == "erc20": row["token_symbol"] = tx.get("tokenSymbol", "") row["token_name"] = tx.get("tokenName", "") row["token_decimal"] = int(tx.get("tokenDecimal", 18)) row["value_token"] = int(tx.get("value", 0)) / (10 ** int(tx.get("tokenDecimal", 18))) rows.append(row) return pd.DataFrame(rows) def normalize_btc_txs(raw_txs: list[dict]) -> pd.DataFrame: """Normalize Bitcoin mempool.space transactions.""" rows = [] for tx in raw_txs: txid = tx.get("txid", "") for vin in tx.get("vin", []): rows.append({ "hash": txid, "from": vin.get("prevout", {}).get("scriptpubkey_address", ""), "to": "", "value_btc": vin.get("prevout", {}).get("value", 0) / 1e8, "type": "input", }) for vout in tx.get("vout", []): rows.append({ "hash": txid, "from": "", "to": vout.get("scriptpubkey_address", ""), "value_btc": vout.get("value", 0) / 1e8, "type": "output", }) return pd.DataFrame(rows) ``` ### Step 3: Build a Directed Transaction Graph Construct a NetworkX directed graph from the normalized transaction data. Each node is an address, each directed edge is a transaction with amount, hash, and timestamp as edge attributes. ```python import networkx as nx from typing import Optional def build_tx_graph(df: pd.DataFrame) -> nx.DiGraph: """Build a directed transaction graph from normalized transaction data. Edges: from → to, attributes include value_eth, hash, timestamp, token_symbol. Self-loops (from == to) are excluded. """ G = nx.DiGraph() for _, row in df.iterrows(): src = str(row.get("from", "")).lower() dst = str(row.get("to", "")).lower() if not src or not dst or src == dst: continue val = row.get("value_eth", 0) or row.get("value_token", 0) or 0 G.add_edge(src, dst, **{ "hash": row.get("hash", ""), "value": float(val), "timestamp": int(row.get("timestamp", 0)), "token": row.get("token_symbol", "ETH"), }) return G def graph_summary(G: nx.DiGraph) -> dict: """Return summary statistics for a transaction graph.""" return { "nodes": G.number_of_nodes(), "edges": G.number_of_edges(), "total_volume": sum(d["value"] for _, _, d in G.edges(data=True)), "unique_sources": sum(1 for n, d in G.in_degree() if d == 0), "unique_sinks": sum(1 for n, d in G.out_degree() if d == 0), "density": nx.density(G), } def filter_by_value(G: nx.DiGraph, min_value: float = 0.01) -> nx.DiGraph: """Return a subgraph containing only edges above a minimum value threshold.""" edges_to_keep = [(u, v, d) for u, v, d in G.edges(data=True) if d.get("value", 0) >= min_value] H = nx.DiGraph() H.add_edges_from(edges_to_keep) return H ``` ### Step 4: Export Graph to DOT and GraphML Generate visualization-ready graph formats for use with Graphviz and analysis tools: ```python def export_dot(G: nx.DiGraph, path: str = "tx_graph.dot") -> str: """Export the transaction graph as a DOT file for Graphviz rendering.""" # Relabel nodes to quoted strings for DOT compatibility H = nx.DiGraph() for u, v, d in G.edges(data=True): label = f"{d.get('value', 0):.4f} {d.get('token', 'ETH')}" H.add_edge(f'"{u[:10]}…"', f'"{v[:10]}…"', label=label) nx.nx_pydot.write_dot(H, path) return path def export_graphml(G: nx.DiGraph, path: str = "tx_graph.graphml") -> str: """Export the transaction graph as GraphML for Gephi or yEd.""" nx.write_graphml(G, path) return path ``` ### Step 5: Compute Graph Centrality and Find Key Nodes Identify the most important addresses in the flow — the sink where funds concentrate, the original source, and intermediary hubs: ```python def find_source_nodes(G: nx.DiGraph) -> list[str]: """Return nodes with zero in-degree (original sources of funds).""" return [n for n, d in G.in_degree() if d == 0] def find_sink_nodes(G: nx.DiGraph) -> list[str]: """Return nodes with zero out-degree (endpoints — mixers, exchanges, bridges).""" return [n for n, d in G.out_degree() if d == 0] def rank_by_centrality(G: nx.DiGraph, top_n: int = 10) -> pd.DataFrame: """Rank addresses by weighted betweenness centrality. Higher centrality means the address sits on more fund flow paths. """ if G.number_of_nodes() < 2: return pd.DataFrame() # Use edge weight (value) to weight paths weight_attr = "value" try: centrality = nx.betweenness_centrality(G, weight=weight_attr) except Exception: centrality = nx.betweenness_centrality(G) ranked = sorted(centrality.items(), key=lambda x: -x[1]) return pd.DataFrame(ranked[:top_n], columns=["address", "centrality"]) def find_top_volume_hubs(G: nx.DiGraph, top_n: int = 10) -> pd.DataFrame: """Find addresses with the highest total transaction volume (in + out).""" volume = {} for u, v, d in G.edges(data=True): val = d.get("value", 0) volume[u] = volume.get(u, 0) + val volume[v] = volume.get(v, 0) + val ranked = sorted(volume.items(), key=lambda x: -x[1]) return pd.DataFrame(ranked[:top_n], columns=["address", "total_value"]) ``` ## Peel Chain Detection Algorithm A peel chain is a money-laundering pattern where a large amount of cryptocurrency is split into gradually smaller amounts through a chain of addresses, obscuring the trail. Each address in the chain typically passes most of the funds forward while peeling off a small amount. ```python def detect_peel_chains(G: nx.DiGraph, max_path_length: int = 20, min_hops: int = 3, value_decay_threshold: float = 0.95) -> list[dict]: """Detect peel chain patterns in a transaction graph. A peel chain is a path where each hop passes most of the value forward (value_decay_threshold controls how much value must be passed), with small amounts peeled off along the way. Returns list of detected chains. """ chains = [] sources = find_source_nodes(G) for src in sources: visited = set() path = [] current = src while current and len(path) < max_path_length: visited.add(current) out_edges = list(G.out_edges(current, data=True)) if not out_edges: break # Sort outgoing edges by value descending out_edges.sort(key=lambda e: e[2].get("value", 0), reverse=True) best = out_edges[0] next_addr = best[1] val = best[2].get("value", 0) # Check: does this edge pass most of the known incoming value? in_edges = list(G.in_edges(current, data=True)) total_in = sum(e[2].get("value", 0) for e in in_edges) if total_in > 0 and val / total_in < (1 - value_decay_threshold): # This hop peeled off too much — likely a consolidation, not a peel chain pass path.append({ "from": current, "to": next_addr, "value": val, "hash": best[2].get("hash", ""), }) if next_addr in visited: break current = next_addr if len(path) >= min_hops: chains.append({ "start": src, "end": current, "hops": len(path), "total_volume": sum(h["value"] for h in path), "path": path, }) return chains def detect_peel_chain_heuristic_simple(address: str, txs: list[dict], threshold_ratio: float = 0.8) -> list[dict]: """Detect peel chains using a simplified heuristic on raw transaction data. A peel chain address sends most of its incoming value to a single output, suggesting gradual fund distribution. This works on individual address analysis without needing a full graph. """ outgoing = {} for tx in txs: frm = tx.get("from", "").lower() to = tx.get("to", "").lower() val = int(tx.get("value", 0)) if frm == address: outgoing[to] = outgoing.get(to, 0) + val if not outgoing: return [] total_out = sum(outgoing.values()) chains = [] for dst, val in sorted(outgoing.items(), key=lambda x: -x[1]): ratio = val / total_out if total_out > 0 else 0 if ratio >= threshold_ratio: chains.append({ "from": address, "to": dst, "value": val, "ratio": round(ratio, 4), "pattern": "peel_chain_hop", }) return chains ``` ## Consolidation Pattern Detector Consolidation is the opposite of a peel chain — multiple addresses send funds to a single address. This is typical before an exchange deposit or when a hacker collects funds spread across many wallets. ```python def detect_consolidation(G: nx.DiGraph, min_sources: int = 3, time_window_hours: int = 24) -> list[dict]: """Detect consolidation patterns in the transaction graph. A consolidation is a node with N incoming edges from distinct addresses within a time window. This is a classic exchange deposit / fund collection pattern. """ consolidations = [] for node in G.nodes(): in_edges = list(G.in_edges(node, data=True)) if len(in_edges) < min_sources: continue unique_sources = set(e[0] for e in in_edges) if len(unique_sources) < min_sources: continue # Check time window timestamps = [e[2].get("timestamp", 0) for e in in_edges] timestamps = [t for t in timestamps if t > 0] if timestamps: span_hours = (max(timestamps) - min(timestamps)) / 3600 if span_hours > time_window_hours: continue total_value = sum(e[2].get("value", 0) for e in in_edges) consolidations.append({ "sink": node, "sources": list(unique_sources), "num_sources": len(unique_sources), "total_value": total_value, "time_window_hours": round(span_hours, 2) if timestamps else 0, "tx_hashes": [e[2].get("hash", "") for e in in_edges], }) return sorted(consolidations, key=lambda c: -c["num_sources"]) def detect_consolidation_simple(txs: list[dict], min_sources: int = 5) -> list[dict]: """Detect consolidation by checking which addresses receive from many distinct senders. Works on raw transaction data without a full graph. """ from collections import Counter receivers = Counter() receiver_txs = {} for tx in txs: to = tx.get("to", "").lower() frm = tx.get("from", "").lower() if to and frm: receivers[to] += 1 if to not in receiver_txs: receiver_txs[to] = [] receiver_txs[to].append(tx) consolidations = [] for addr, count in receivers.most_common(): if count >= min_sources: total_val = sum(int(tx.get("value", 0)) for tx in receiver_txs[addr]) consolidations.append({ "address": addr, "incoming_count": count, "total_value": total_val, "source_addresses": list(set( tx.get("from", "").lower() for tx in receiver_txs[addr] )), }) return consolidations ``` ## Mixer Identification ### Known Mixer/Sanitizer Contract Addresses | Service | Chain | Address | Type | Notes | |---|---|---|---|---| | Tornado Cash | Ethereum | `0x12d66f87a04a9e220743712ce6d9bb1b5616b8fc` | ETH mixer | 0.1 ETH pool | | Tornado Cash | Ethereum | `0x47ce0c6ed5b0ce3d3a51fdb1c52dc66a7c3c2936` | ETH mixer | 1 ETH pool | | Tornado Cash | Ethereum | `0x910cbd523d972eb0a6f4cae4618ad62622b39dbf` | ETH mixer | 10 ETH pool | | Tornado Cash | Ethereum | `0xa160cdab2250da5b3f80029b6b82e1d5d0c30efb` | ETH mixer | 100 ETH pool | | Tornado Cash | BSC | `0x1e34a77868e19a6647b1f2f47b51ed72dede95dd` | BNB mixer | 100 BNB pool | | Tornado Cash | Polygon | `0x47ce0c6ed5b0ce3d3a51fdb1c52dc66a7c3c2936` | MATIC mixer | 1000 MATIC pool | | Sinbad | Ethereum | `0x1e31cb9b6b69a7df3225220a0ef1f0a76586d5a3` | ETH mixer | | | Sinbad | Ethereum | `0x8589427373d6d84e98730d7795d8f6f8731fda16` | ETH mixer | | | Wasabi Wallet | Bitcoin | `bc1q…` (varies) | BTC CoinJoin | Uses PayNym, no fixed address | | Samourai Whirlpool | Bitcoin | (Coordinator) | BTC CoinJoin | No fixed deposit address | | ChainFlip | Ethereum | `0x60fB0B27447480E3304bB7b2660e9C081D3775fc` | Cross-chain mixer | | | eXch | Ethereum | (varies) | Swap-based mixer | Swaps and splits | | FixedFloat | Multi | (varies) | Swap-based mixer | No-KYC exchange used for obfuscation | ### Mixer Deposit Detection Heuristics ```python # Known mixer contract addresses (Ethereum mainnet) MIXER_ADDRESSES = { # Tornado Cash deposit addresses (by pool size) "0x12d66f87a04a9e220743712ce6d9bb1b5616b8fc": "tornado_0.1_eth", "0x47ce0c6ed5b0ce3d3a51fdb1c52dc66a7c3c2936": "tornado_1_eth", "0x910cbd523d972eb0a6f4cae4618ad62622b39dbf": "tornado_10_eth", "0xa160cdab2250da5b3f80029b6b82e1d5d0c30efb": "tornado_100_eth", # Sinbad "0x1e31cb9b6b69a7df3225220a0ef1f0a76586d5a3": "sinbad_1", "0x8589427373d6d84e98730d7795d8f6f8731fda16": "sinbad_2", } def tag_mixer_interactions(df: pd.DataFrame) -> pd.DataFrame: """Tag transactions involving known mixer addresses.""" df_lower = df.copy() df_lower["to"] = df_lower["to"].str.lower() df_lower["from"] = df_lower["from"].str.lower() df_lower["mixer"] = "unknown" for addr, name in MIXER_ADDRESSES.items(): mask = (df_lower["to"] == addr) | (df_lower["from"] == addr) df_lower.loc[mask, "mixer"] = name return df_lower def detect_mixer_deposit_by_amount(address: str, txs: list[dict], known_amounts: list[float] = None) -> bool: """Detect potential mixer deposits by checking for exact-amount patterns. Mixers use fixed deposit amounts (e.g., Tornado Cash pools). If an address sends an exact pool amount to a contract, it's likely a mixer deposit. """ if known_amounts is None: known_amounts = [0.1, 1.0, 10.0, 100.0] # Tornado Cash standard pool amounts for tx in txs: frm = tx.get("from", "").lower() val_eth = int(tx.get("value", 0)) / 1e18 if frm == address.lower() and val_eth in known_amounts: # Verify the recipient has contract code (mixer contract, not EOA) return True return False def detect_mixer_timing_anomaly(txs: list[dict], time_window_minutes: int = 30) -> list[dict]: """Detect mixer deposit patterns by timing. Mixer deposits often cluster in tight time windows. Multiple deposits to different addresses from the same source within a short window is a strong mixer indicator. """ from collections import defaultdict windows = defaultdict(list) for tx in txs: ts = int(tx.get("timeStamp", 0)) to = tx.get("to", "").lower() if ts: bucket_key = ts // (time_window_minutes * 60) # bucket by time window windows[bucket_key].append(tx) suspicious = [] for bucket, group in windows.items(): unique_recipients = set(t.get("to", "").lower() for t in group) if len(unique_recipients) >= 3: total_val = sum(int(t.get("value", 0)) for t in group) / 1e18 suspicious.append({ "time_bucket": bucket, "num_txs": len(group), "unique_recipients": len(unique_recipients), "total_value_eth": total_val, "txs": [t.get("hash", "") for t in group[:10]], }) return suspicious ``` ## Cross-Chain Bridge Tracing Bridges are the most common obfuscation technique post-hack. The attacker deposits on Chain A and withdraws on Chain B, often swapping the asset type. ### Known Bridge Contract Addresses | Bridge | Chain | Contract Address | Event Signature | |---|---|---|---| | LayerZero Endpoint | Ethereum | `0x66A71Dcef29A0fFBDBE3c6a460a3B5BC225Cd675` | `MessagePayload` | | Stargate Router | Ethereum | `0x8731d54E9D02c286767d56ac03e8037C07e01e98` | `Swap` | | Stargate Router | Arbitrum | `0x53Bf833A5d6c4ddA888F69c22C88C9f356a41614` | `Swap` | | Wormhole Core | Ethereum | `0x98f3c9e6E3fAce36bAAd05FE09d375Ef1464288B` | `LogMessagePublished` | | Wormhole Token Bridge | Ethereum | `0x3ee18B2214AFF97000D974cf647E7C347E8fa585` | `TransferRedeemed` | | Across | Ethereum | `0x5c7BCd6E7De5423a257D81B442095A1a6ced35C5` | `FundsDeposited` | | Across | Arbitrum | `0x3bB4445D30AC020a84c1b5A8A2C6248ebC9779D0` | `FundsDeposited` | | Synapse | Ethereum | `0x2796317b0fF8538F253012862c06787Adfb8c3aC` | `TokenDeposit` | | Synapse | Arbitrum | `0x9D33eeE1540BdA15C2021C22E14dfdB41c58485b` | `TokenDeposit` | | Hop | Ethereum | `0x3666f603Cc164936C1b87e207F36BEBa4AC5f18a` | `TransferSent` | | Hop | Polygon | `0x3666f603Cc164936C1b87e207F36BEBa4AC5f18a` | `TransferSent` | | AnySwap (Multichain) | Ethereum | `0x6b7a87899490EcE95443e979cA9485CBE7E71522` | `AnySwapOut` | | Celer cBridge | Ethereum | `0x1619DE6B6B20eD217a58d00f37B9d47C7663feca` | `Send` | ### Bridge Detection Code ```python BRIDGE_CONTRACTS = { "layerzero_endpoint": { "ethereum": ["0x66a71dcef29a0ffbdbe3c6a460a3b5bc225cd675"], "arbitrum": ["0x3c2269811836af69497e5f486a85d7316753cf02"], "polygon": ["0x3c2269811836af69497e5f486a85d7316753cf02"], }, "stargate_router": { "ethereum": ["0x8731d54e9d02c286767d56ac03e8037c07e01e98"], "arbitrum": ["0x53bf833a5d6c4dda888f69c22c88c9f356a41614"], "polygon": ["0x53bf833a5d6c4dda888f69c22c88c9f356a41614"], }, "across": { "ethereum": ["0x5c7bcd6e7de5423a257d81b442095a1a6ced35c5"], "arbitrum": ["0x3bb4445d30ac020a84c1b5a8a2c6248ebc9779d0"], }, "synapse": { "ethereum": ["0x2796317b0ff8538f253012862c06787adfb8c3ac"], "arbitrum": ["0x9d33eee1540bda15c2021c22e14dfdb41c58485b"], }, "wormhole": { "ethereum": ["0x98f3c9e6e3face36baad05fe09d375ef1464288b"], }, "hop": { "ethereum": ["0x3666f603cc164936c1b87e207f36beba4ac5f18a"], "polygon": ["0x3666f603cc164936c1b87e207f36beba4ac5f18a"], }, "celer": { "ethereum": ["0x1619de6b6b20ed217a58d00f37b9d47c7663feca"], }, } # Flatten bridge addresses to a single set for fast lookup ALL_BRIDGE_ADDRESSES = set() for name, chains in BRIDGE_CONTRACTS.items(): for chain_addrs in chains.values(): ALL_BRIDGE_ADDRESSES.update(chain_addrs) def tag_bridge_interactions(df: pd.DataFrame) -> pd.DataFrame: """Tag transactions involving known bridge contracts.""" df_lower = df.copy() for col in ["to", "from"]: df_lower[col] = df_lower[col].str.lower() def lookup(addr): addr = addr.lower() for name, chains in BRIDGE_CONTRACTS.items(): for chain_addrs in chains.values(): if addr in chain_addrs: return name return "" df_lower["bridge"] = df_lower["to"].apply(lookup) # Also check 'from' field for bridge interactions mask_from = df_lower["bridge"] == "" df_lower.loc[mask_from, "bridge"] = df_lower.loc[mask_from, "from"].apply(lookup) return df_lower def detect_bridge_transfer(tx: dict, receipt_logs: list[dict]) -> dict: """Detect bridge transfer from a transaction's event logs. Returns destination chain info if a bridge event is found. """ for log in receipt_logs: addr = log.get("address", "").lower() topics = log.get("topics", []) # Stargate Swap event: topics[0] = Swap(uint16,uint16,address,address,uint256,uint256) if addr in ALL_BRIDGE_ADDRESSES and len(topics) >= 3: return { "bridge_detected": True, "contract": addr, "tx_hash": tx.get("hash", ""), "topics": topics, "data": log.get("data", ""), "note": "Bridge interaction detected — check destination chain in event data", } return {"bridge_detected": False} def extract_bridge_destination(tx_hash: str, web3_provider, bridge_address: str) -> int: """Extract destination chain ID from a Stargate bridge event. Stargate emits Swap(uint16 chainIdFrom, uint16 chainIdTo, ...). The chainIdTo is typically topic[2] in the Swap event. """ receipt = web3_provider.eth.get_transaction_receipt(tx_hash) for log in receipt.logs: if log.address.lower() == bridge_address.lower(): topics = log.topics if len(topics) >= 3: dest_chain_id = int.from_bytes(topics[2], "big") return dest_chain_id return 0 ``` ## Wallet Clustering Implementation ### CoJoin / Common-Input Clustering (UTXO Chains) Bitcoin-like UTXO chains enable a powerful clustering heuristic: if two addresses appear as inputs in the same transaction, they are controlled by the same entity (assuming multi-signature or CoinJoin is not involved). This is called CoJoin or Common-Input-Ownership Heuristic. ```python def cluster_by_common_input(txs_utxo: list[dict]) -> dict[str, set[str]]: """Cluster Bitcoin addresses by common-input ownership heuristic. If addresses A and B appear as inputs in the same transaction, they are likely controlled by the same entity. Returns a dict mapping cluster_id -> set of addresses. """ address_to_tx = {} tx_to_addresses = {} for tx in txs_utxo: txid = tx.get("txid", "") input_addresses = set() for vin in tx.get("vin", []): addr = vin.get("prevout", {}).get("scriptpubkey_address", "") if addr: input_addresses.add(addr) if len(input_addresses) >= 2: tx_to_addresses[txid] = input_addresses for addr in input_addresses: if addr not in address_to_tx: address_to_tx[addr] = [] address_to_tx[addr].append(txid) # Union-Find to merge clusters parent = {} def find(x): while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def union(a, b): ra, rb = find(a), find(b) if ra != rb: parent[ra] = rb all_addresses = set() for addresses in tx_to_addresses.values(): addrs = list(addresses) for addr in addrs: if addr not in parent: parent[addr] = addr all_addresses.add(addr) for i in range(1, len(addrs)): union(addrs[0], addrs[i]) # Build clusters clusters = {} for addr in all_addresses: root = find(addr) if root not in clusters: clusters[root] = set() clusters[root].add(addr) return clusters def cluster_by_deposit_address(txs: list[dict], exchange_pattern: str = "binance") -> list[dict]: """Identify exchange deposit addresses by known patterns. Many exchanges use deterministic deposit addresses. This heuristic clusters addresses that receive from many senders as potential exchange deposit addresses. """ from collections import Counter, defaultdict deposits_to = Counter() depositors = defaultdict(set) for tx in txs: to = tx.get("to", "").lower() frm = tx.get("from", "").lower() if to and frm: deposits_to[to] += 1 depositors[to].add(frm) clusters = [] for addr, count in deposits_to.most_common(50): if count >= 10: clusters.append({ "deposit_address": addr, "incoming_count": count, "unique_depositors": len(depositors[addr]), "likely_entity": exchange_pattern if count > 50 else "unknown", }) return clusters ``` ### EVM Address Reuse Clustering Even on account-based chains, addresses can be clustered by behavioral patterns: ```python def cluster_by_behavior(df: pd.DataFrame, min_interactions: int = 5) -> dict[str, list[str]]: """Cluster EVM addresses by behavioral similarity. If two addresses interact with the same set of contracts/protocols, they may be controlled by the same entity. """ from collections import defaultdict addr_contracts = defaultdict(set) for _, row in df.iterrows(): frm = str(row.get("from", "")).lower() to = str(row.get("to", "")).lower() if frm and to: addr_contracts[frm].add(to) if to: addr_contracts[to] # ensure key exists # Jaccard similarity between address contract sets addrs = list(addr_contracts.keys()) clusters = defaultdict(list) assigned = set() for i, a1 in enumerate(addrs): if a1 in assigned: continue set1 = addr_contracts[a1] if len(set1) < min_interactions: continue cluster = [a1] assigned.add(a1) for a2 in addrs[i + 1:]: if a2 in assigned: continue set2 = addr_contracts[a2] if len(set2) < min_interactions: continue intersection = len(set1 & set2) union = len(set1 | set2) if union > 0 and intersection / union >= 0.7: cluster.append(a2) assigned.add(a2) if len(cluster) >= 2: clusters[f"behavioral_cluster_{len(clusters)}"] = cluster r …(truncated)