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

Etherscan-like API Wrapper

The following chains use an identical API schema to Etherscan. A single helper class handles all of them:

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:

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:

Bitcoin Transaction Fetcher (mempool.space)

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

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.

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:

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:

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.

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:

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:

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.

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.

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

Mixer Deposit Detection Heuristics

# 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 Detection Code

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.

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:

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)