oyi77 Wallet & Address Intelligence
Overview
Wallet address intelligence is the practice of profiling blockchain addresses to identify the entities behind them, assess risk levels, and build behavioral profiles. Unlike simple balance checks, address intelligence combines on-chain transaction analysis, entity clustering heuristics, known-address databases, behavioral pattern matching, and cross-chain correlation. This skill covers building address profiles with transaction demographics, applying clustering algorithms (Cointracking, CommonInput, behavioral similarity), tagging addresses with entity attributions, assessing risk scores for compliance (AML/KYC), sanctions list screening, MEV searcher identification, ENS/social graph resolution, and producing comprehensive address intelligence reports suitable for compliance teams, security analysts, and protocol operators.
When to Use
Trigger phrases:
"wallet address intelligence"
"Profile a blockchain wallet"
"Check if an address is risky"
"Cluster related crypto wallets"
"Build address reputation"
"Run AML screening on an address"
"Find all addresses controlled by the same entity"
"Analyze wallet interaction patterns"
When performing due diligence on a counterparty before a transaction
When investigating suspicious addresses for compliance or AML purposes
When building a wallet clustering system for forensic investigations
When assessing the risk profile of an address interacting with your protocol
When attributing addresses to known entities (exchanges, protocols, exploiters)
When screening addresses against sanctions lists before onboarding
When linking addresses across different blockchains for entity tracking
When identifying MEV bots and searcher strategies
When generating address intelligence reports for law enforcement or compliance
When building a watchlist monitoring system for high-risk addresses
When NOT to Use
- You need to trace stolen funds along a specific path (use onchain-transaction-forensics skill)
- You need to analyze a specific DeFi protocol hack (use defi-incident-analysis skill)
- You need to check if a token is a scam (use token-nft-scam-investigation skill)
- You need real-time transaction monitoring (use a blockchain analytics platform like Chainalysis or Elliptic)
- The address is on a privacy-focused blockchain where clustering is infeasible (Monero, Zcash with shielded transactions)
- You only need a single balance check (use a block explorer or a simple RPC call)
- The target uses a privacy wallet with coin-join or stealth addresses
Prerequisites
- Python 3.8+ with web3.py, requests, pandas, networkx, numpy, jellyfish (fuzzy matching)
- Block explorer API keys (Etherscan, BscScan, Polygonscan, etc.)
- Access to an address labeling dataset (Etherscan labels, tagged addresses from block explorers, known exploiters list)
- Basic understanding of graph theory for clustering algorithms
- Optional: DeBank or similar API for multi-chain wallet views
- Optional: Flashbots API access for MEV bundle data
- Optional: ENS resolution library (web3.py handles this natively)
pip install web3 requests pandas networkx numpy jellyfish
Core Workflow
The core workflow builds a comprehensive address intelligence pipeline: collect data, cluster related addresses, score risk, attribute entities, and produce a report.
Address Profile Builder
import requests
from datetime import datetime, timezone
from typing import Optional
import json
# ============================================================
# BLOCK EXPLORER HELPERS
# ============================================================
ETHERSCAN_BASE = "https://api.etherscan.io/api"
def _fetch_balance(address: str, api_key: str) -> float:
"""Fetch ETH balance from Etherscan."""
params = {"module": "account", "action": "balance", "address": address, "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=15)
data = resp.json()
if data.get("status") == "1":
return int(data["result"]) / 1e18
return 0.0
def _fetch_tx_list(address: str, api_key: str, startblock: int = 0, endblock: int = 99999999) -> list:
"""Fetch normal transaction list for an address."""
params = {"module": "account", "action": "txlist", "address": address,
"startblock": startblock, "endblock": endblock, "sort": "desc", "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=30)
data = resp.json()
return data.get("result", [])
def _fetch_internal_tx_list(address: str, api_key: str) -> list:
"""Fetch internal transactions for an address."""
params = {"module": "account", "action": "txlistinternal", "address": address,
"startblock": 0, "endblock": 99999999, "sort": "desc", "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=30)
data = resp.json()
return data.get("result", [])
def _fetch_erc20_transfers(address: str, api_key: str) -> list:
"""Fetch ERC-20 token transfer events."""
params = {"module": "account", "action": "tokentx", "address": address,
"startblock": 0, "endblock": 99999999, "sort": "desc", "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=30)
data = resp.json()
return data.get("result", [])
def _fetch_nft_transfers(address: str, api_key: str) -> list:
"""Fetch NFT (ERC-721 / ERC-1155) transfer events."""
params = {"module": "account", "action": "tokennfttx", "address": address,
"startblock": 0, "endblock": 99999999, "sort": "desc", "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=30)
data = resp.json()
return data.get("result", [])
def _fetch_token_balances(address: str, api_key: str) -> list:
"""Fetch all token balances for an address."""
params = {"module": "account", "action": "tokenlist", "address": address, "apikey": api_key}
resp = requests.get(ETHERSCAN_BASE, params=params, timeout=15)
data = resp.json()
return data.get("result", [])
def build_address_profile(address: str, etherscan_key: str) -> dict:
"""Build a comprehensive profile for a single address."""
profile = {
"address": address,
"tags": [],
"risk_indicators": [],
"activity": {},
"known_labels": [],
"token_holdings": [],
"nft_holdings": [],
"protocols_used": set(),
"interactors": set(),
"gas_profile": {},
}
# 1. Basic balance and transaction count
balance = _fetch_balance(address, etherscan_key)
tx_list = _fetch_tx_list(address, etherscan_key)
internal_txs = _fetch_internal_tx_list(address, etherscan_key)
erc20_txs = _fetch_erc20_transfers(address, etherscan_key)
nft_txs = _fetch_nft_transfers(address, etherscan_key)
profile["eth_balance"] = balance
profile["total_txns"] = len(tx_list)
profile["total_internal_txns"] = len(internal_txs)
profile["total_token_transfers"] = len(erc20_txs)
profile["total_nft_transfers"] = len(nft_txs)
profile["total_activity"] = len(tx_list) + len(internal_txs) + len(erc20_txs)
# 2. First and last activity
all_txs_sorted = sorted(tx_list, key=lambda t: int(t.get("timeStamp", 0)), reverse=True)
if all_txs_sorted:
profile["first_seen"] = datetime.fromtimestamp(int(all_txs_sorted[-1]["timeStamp"]), tz=timezone.utc)
profile["last_active"] = datetime.fromtimestamp(int(all_txs_sorted[0]["timeStamp"]), tz=timezone.utc)
profile["age_days"] = (profile["last_active"] - profile["first_seen"]).days if profile["first_seen"] else 0
# 3. Interaction diversity — unique addresses this address has interacted with
unique_interactors = set()
protocols = set()
gas_prices = []
for tx in all_txs_sorted:
unique_interactors.add(tx["from"].lower())
if tx["to"] and tx["to"] != address.lower():
unique_interactors.add(tx["to"].lower())
protocols.add(tx["to"].lower())
if tx.get("gasPrice"):
gas_prices.append(int(tx["gasPrice"]))
profile["unique_interactors"] = len(unique_interactors)
profile["protocols_touched"] = len(protocols)
# Gas profile
if gas_prices:
profile["gas_profile"] = {
"min_gwei": min(gas_prices) / 1e9,
"max_gwei": max(gas_prices) / 1e9,
"avg_gwei": (sum(gas_prices) / len(gas_prices)) / 1e9,
"total_gas_spent_eth": sum(
int(tx.get("gasUsed", 0)) * int(tx.get("gasPrice", 0)) / 1e18
for tx in all_txs_sorted if tx.get("gasUsed")
),
}
# 4. Token holdings
token_balances = _fetch_token_balances(address, etherscan_key)
profile["token_holdings"] = [
{
"token": t.get("tokenName", "unknown"),
"symbol": t.get("tokenSymbol", "???"),
"contract": t.get("contractAddress", "").lower(),
"balance": float(t.get("balance", 0)) / 10 ** int(t.get("tokenDecimal", 18)),
}
for t in token_balances if t.get("balance") and int(t.get("balance", 0)) > 0
]
# 5. Known entity labels
labels = _check_known_labels(address)
if labels:
profile["known_labels"].extend(labels)
profile["tags"].extend(labels)
return profile
def _check_known_labels(address: str) -> list:
"""Check address against hardcoded known-entity databases (Etherscan labels, known exploiters, exchanges).
In production, load from a database or JSON file rather than hardcoding.
"""
address_lower = address.lower()
# Known exchange deposit addresses (subset for illustration)
known_exchanges = {
"0x3f5ce5fbfe3e9af3971dd833d26ba9b5c936f0be": "Binance 1",
"0xd551234ae421e3bcba99a0da6d736074f22192ff": "Binance 2",
"0x28c6c06298d514db089934071355e5743bf21d60": "Binance 3",
"0x21a31ee1afc51d94c2efccaa2092ad1028285549": "Coinbase 1",
"0xa090e606e30bd747d4e6245a1517ebe430f0057e": "Coinbase 2",
"0xe92d1a43df510ff82d2218671ae56b150287a6f5": "Kraken 1",
"0x0a869d79a7052c7f1b55a8ebabbea3420f0d1e13": "Kraken 2",
"0x126783cba8df91c1c42ee59d1cfb342f42f04ce3": "KuCoin",
"0x281dc6b700385c8e826a7e0c1f6b7e10f5f2d894": "Bitfinex",
}
# Known exploiter / hacker addresses (subset)
known_exploiters = {
"0x1e227979f6b5c9704f9a92e4201ffc7d7c2d7bbf": "Bybit Exploiter (Lazarus)",
"0x59e0cda5922ef1a80d49f5fe4714e2343dd2ae4f": "Ronin Bridge Exploiter",
"0x098b716b8aaf21512996dc57eb0615e2383e2f96": "Nomad Bridge Exploiter",
"0x0de8f4f3c92abb2fc3a6c4ad07a39bbffa4c37a5": "Wormhole Exploiter",
"0x5dafb0d0f71b5acd3a1d4e21a358e7dcb75bceff": "FTX Drainer",
}
# Known darknet market deposit / payout addresses (subset)
known_darknet = {
"0x5f4ec3df9cd534eda9715a9fa20f80283c8c48be": "Hydra Market",
"0x1548d173e0f9d2b9b46a69e3f8f1c1c1f9b9c0a1": "AlphaBay (seized)",
}
# Known ransomware payment / actor-controlled addresses (subset)
known_ransomware = {
"0x9e39b3c92a3f6a2c8d0b6e9f0c4a7b3d2e1f0a9b": "Conti Ransomware",
"0x3c4a1b7e9f2d8c5a6b0e3f1d4c7a9b2e5f0d3c8a": "LockBit Ransomware",
}
# CEX operating from OFAC-sanctioned jurisdictions.
# Do NOT hardcode addresses here — load the OFAC SDN list via load_sanctions_list().
# (Coinbase entries were incorrectly listed as sanctioned; removed to avoid false positives.)
known_sanctioned_cex = {}
# Mixers
known_mixers = {
"0x910cbd523d972eb0a6f4cae4618ad62622b39dbf": "Tornado Cash 1",
"0xa160cdab225685da1d56aa342ad8841c3b53f291": "Tornado Cash 2",
"0x12d66f87a04a9e220743712ce6d9bb1b5616b8fc": "Tornado Cash 3",
"0x47ce0c6ed5b0ce3d3a51fdb1c52dc66a7c3c2936": "Tornado Cash 4",
}
results = []
if address_lower in known_exchanges:
results.append(("exchange", known_exchanges[address_lower]))
if address_lower in known_exploiters:
results.append(("exploiter", known_exploiters[address_lower]))
if address_lower in known_mixers:
results.append(("mixer", known_mixers[address_lower]))
if address_lower in known_darknet:
results.append(("darknet", known_darknet[address_lower]))
if address_lower in known_ransomware:
results.append(("ransomware", known_ransomware[address_lower]))
if address_lower in known_sanctioned_cex:
results.append(("sanctioned_cex", known_sanctioned_cex[address_lower]))
return results
# ============================================================
# COMMON-INPUT CLUSTERING (Bitcoin-style)
# ============================================================
def build_common_input_clusters(tx_inputs: list[list[str]]) -> dict:
"""Cluster Bitcoin addresses that appear together as inputs in the same transaction.
This is the classic CommonInput heuristic used by Chainalysis and similar tools:
if two addresses are inputs to the same transaction, they are controlled by the same entity
(because only the entity controlling both signing keys would choose to spend from both in one tx).
Args:
tx_inputs: List of lists, where each inner list contains input addresses for one transaction.
Returns:
Dict mapping cluster_id (int) to a set of addresses in that cluster.
Example:
>>> txs = [
... ["1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa", "1BvBMSEYstWetqTFn5Au4m4GFg7xJaNVN2"],
... ["1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa", "1C5e8CzVX6TGhP1qYLPr5XzL4yZLj9Mq6S"],
... ["1DqQn5YpP2MjoBRKJoTQ9rTmW3JfLQZtHx"],
... ]
>>> cl = build_common_input_clusters(txs)
>>> # "1A1z..." and "1BvBM..." and "1C5e..." are in the same cluster
"""
parent = {}
cluster_id_counter = [0]
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(x, y):
rx, ry = find(x), find(y)
if rx != ry:
parent[ry] = rx
for inputs in tx_inputs:
if len(inputs) < 2:
continue # Single-input tx adds no clustering info
# Ensure all addresses have entries in union-find
for addr in inputs:
if addr not in parent:
parent[addr] = addr
# Union all pairs — they are controlled by the same entity
first = inputs[0]
for addr in inputs[1:]:
union(first, addr)
# Collect clusters
clusters = {}
for addr in parent:
root = find(addr)
if root not in clusters:
clusters[root] = set()
clusters[root].add(addr)
# Assign numeric IDs
result = {}
for idx, (root, members) in enumerate(clusters.items()):
result[idx] = members
return result
def cluster_stats(clusters: dict) -> dict:
"""Return summary statistics for clustered address groups."""
member_counts = [len(members) for members in clusters.values()]
return {
"total_clusters": len(clusters),
"total_addresses": sum(member_counts),
"avg_cluster_size": sum(member_counts) / len(member_counts) if member_counts else 0,
"largest_cluster": max(member_counts) if member_counts else 0,
"singleton_clusters": sum(1 for c in member_counts if c == 1),
}
# ============================================================
# BEHAVIORAL SIMILARITY CLUSTERING
# ============================================================
def compute_behavioral_similarity(profiles: list[dict]) -> list[tuple[int, int, float]]:
"""Compute pairwise similarity scores between address profiles based on behavioral patterns.
Similarity dimensions:
- Protocol touch overlap (Jaccard similarity of contracts interacted with)
- Time-of-day activity distribution similarity (cosine)
- Gas price preferences (mean gas price delta)
- Number of unique interactors (ratio-based)
Returns list of (profile_a_index, profile_b_index, similarity_score) tuples,
where similarity_score is in [0, 1].
"""
from sklearn.feature_extraction.text import TfidfVectorizer # optional improvement
import numpy as np
import math
n = len(profiles)
scores = []
for i in range(n):
for j in range(i + 1, n):
p1 = profiles[i]
p2 = profiles[j]
dims = []
# 1. Protocol overlap (Jaccard)
prots1 = p1.get("protocols_touched", set()) if isinstance(p1.get("protocols_touched"), set) else set()
prots2 = p2.get("protocols_touched", set()) if isinstance(p2.get("protocols_touched"), set) else set()
if prots1 or prots2:
intersection = len(prots1 & prots2)
union = len(prots1 | prots2)
jaccard = intersection / union if union > 0 else 0
dims.append(jaccard)
# 2. Total activity volume similarity (ratio-based)
act1 = p1.get("total_activity", 0)
act2 = p2.get("total_activity", 0)
if act1 > 0 or act2 > 0:
vol_sim = min(act1, act2) / max(act1, act2) if max(act1, act2) > 0 else 0
dims.append(vol_sim)
# 3. Wallet age similarity
age1 = p1.get("age_days", 0)
age2 = p2.get("age_days", 0)
if age1 > 0 or age2 > 0:
age_sim = min(age1, age2) / max(age1, age2) if max(age1, age2) > 0 else 0
dims.append(age_sim)
# 4. Gas price preference similarity
gp1 = p1.get("gas_profile", {}).get("avg_gwei", 0)
gp2 = p2.get("gas_profile", {}).get("avg_gwei", 0)
if gp1 > 0 and gp2 > 0:
gas_sim = min(gp1, gp2) / max(gp1, gp2) if max(gp1, gp2) > 0 else 0
dims.append(gas_sim)
if dims:
combined = sum(dims) / len(dims)
scores.append((i, j, round(combined, 4)))
return scores
def cluster_by_behavior(scores: list[tuple[int, int, float]], threshold: float = 0.7) -> list[list[int]]:
"""Group address indices into behavior-based clusters using threshold on similarity scores.
Args:
scores: Output from compute_behavioral_similarity.
threshold: Minimum similarity to consider two addresses behaviorally linked (default 0.7).
Returns:
List of clusters, each cluster is a list of address indices.
"""
parent = {}
def find(x):
while parent[x] != x:
parent[x] = parent[parent[x]]
x = parent[x]
return x
def union(x, y):
rx, ry = find(x), find(y)
if rx != ry:
parent[ry] = rx
for i, j, sim in scores:
if sim >= threshold:
if i not in parent:
parent[i] = i
if j not in parent:
parent[j] = j
union(i, j)
# Collect clusters
clusters_map = {}
for idx in parent:
root = find(idx)
if root not in clusters_map:
clusters_map[root] = []
clusters_map[root].append(idx)
return list(clusters_map.values())
# ============================================================
# INTERACTION NETWORK BUILDER (NetworkX)
# ============================================================
import networkx as nx
def build_interaction_network(tx_list: list[dict], min_interactions: int = 1) -> nx.DiGraph:
"""Build a weighted directed graph of address interactions from transaction data.
Nodes = addresses (EOA + contracts).
Edges = directional value flow. Weight = number of transactions between the pair.
Args:
tx_list: List of transaction dicts (from Etherscan txlist).
min_interactions: Minimum txs between a pair to include the edge (filter noise).
Returns:
NetworkX DiGraph with node and edge attributes.
"""
G = nx.DiGraph()
for tx in tx_list:
sender = tx.get("from", "").lower()
receiver = tx.get("to", "").lower()
value_eth = int(tx.get("value", 0)) / 1e18
if not sender or not receiver:
continue
G.add_node(sender, node_type="sender")
G.add_node(receiver, node_type="receiver")
if G.has_edge(sender, receiver):
G[sender][receiver]["weight"] += 1
G[sender][receiver]["total_value"] += value_eth
else:
G.add_edge(sender, receiver, weight=1, total_value=value_eth)
# Remove low-weight edges
edges_to_remove = [(u, v) for u, v, d in G.edges(data=True) if d["weight"] < min_interactions]
G.remove_edges_from(edges_to_remove)
return G
def compute_network_stats(G: nx.DiGraph) -> dict:
"""Compute centrality and structure metrics for an interaction network.
Returns dict with:
- pagerank: Dict of node -> PageRank score
- in_degree_centrality: Dict of node -> in-degree centrality
- out_degree_centrality: Dict of node -> out-degree centrality
- betweenness_centrality: Top-10 nodes by betweenness
- density: Network density (0-1)
- strongly_connected_components: Number of SCCs
"""
stats = {}
if G.number_of_nodes() == 0:
return {"error": "empty graph"}
try:
stats["pagerank"] = nx.pagerank(G, alpha=0.85)
except nx.PowerIterationFailedConvergence:
stats["pagerank"] = {}
stats["in_degree_centrality"] = nx.in_degree_centrality(G)
stats["out_degree_centrality"] = nx.out_degree_centrality(G)
# Betweenness — compute only for top nodes due to O(n^3) complexity
if G.number_of_nodes() < 1000:
betweenness = nx.betweenness_centrality(G, k=min(50, G.number_of_nodes()))
top_betweenness = sorted(betweenness.items(), key=lambda x: x[1], reverse=True)[:10]
stats["top_betweenness"] = top_betweenness
else:
stats["top_betweenness"] = []
stats["density"] = nx.density(G)
stats["nodes"] = G.number_of_nodes()
stats["edges"] = G.number_of_edges()
try:
stats["scc_count"] = nx.number_strongly_connected_components(G)
except Exception:
stats["scc_count"] = 0
return stats
def find_central_hubs(G: nx.DiGraph, top_n: int = 10) -> list[tuple[str, float]]:
"""Find the most central addresses by PageRank score."""
pr = nx.pagerank(G, alpha=0.85)
return sorted(pr.items(), key=lambda x: x[1], reverse=True)[:top_n]
# ============================================================
# KNOWN ADDRESS LABELING
# ============================================================
def load_known_addresses(labels_file: str) -> dict:
"""Load a known-address database from a JSON file.
Expected format:
{
"0x...": {"label": "Binance Hot Wallet", "category": "exchange", "confidence": 0.95},
...
}
"""
with open(labels_file, "r") as f:
return json.load(f)
def label_addresses(addresses: list[str], known_db: dict) -> dict[str, list[dict]]:
"""Tag a list of addresses with labels from the known-address database.
Returns dict mapping address -> list of matched labels.
"""
results = {}
for addr in addresses:
addr_lower = addr.lower()
if addr_lower in known_db:
results[addr] = [known_db[addr]]
else:
results[addr] = []
return results
# ============================================================
# RISK SCORING ENGINE
# ============================================================
RISK_DIMENSIONS = [
"sanctions_match",
"mixer_interaction",
"exploit_association",
"phishing_association",
"darknet_market",
"ransomware_payment",
"flashloan_abuse",
"wash_trading",
"cex_sanctioned_jurisdiction",
]
# Default weights for each risk dimension (sum = 1.0)
DEFAULT_RISK_WEIGHTS = {
"sanctions_match": 0.25,
"mixer_interaction": 0.15,
"exploit_association": 0.20,
"phishing_association": 0.12,
"darknet_market": 0.10,
"ransomware_payment": 0.10,
"flashloan_abuse": 0.03,
"wash_trading": 0.03,
"cex_sanctioned_jurisdiction": 0.02,
}
def compute_risk_score(profile: dict, address: str, weights: dict = None) -> dict:
"""Compute a multi-dimensional risk score for an address.
Each dimension scores 0.0 (no risk) to 1.0 (maximum risk).
The overall score is a weighted sum across dimensions.
Args:
profile: Address profile dict from build_address_profile().
address: The target address.
weights: Dict of dimension -> weight. Uses DEFAULT_RISK_WEIGHTS if None.
Returns:
Dict containing per-dimension scores, overall score (0-100), and confidence level.
"""
if weights is None:
weights = DEFAULT_RISK_WEIGHTS
scores = {}
address_lower = address.lower()
evidence = []
# 1. Sanctions match
sanctions_list = load_sanctions_list() if False else {} # Loaded lazily in practice
has_sanctions = any(
label[0] == "sanctioned" for label in profile.get("known_labels", [])
)
scores["sanctions_match"] = 1.0 if has_sanctions else 0.0
if has_sanctions:
evidence.append("Address matches known sanctions list entry")
# 2. Mixer interaction
has_mixer = any(
label[0] == "mixer" for label in profile.get("known_labels", [])
)
scores["mixer_interaction"] = 1.0 if has_mixer else 0.0
if has_mixer:
evidence.append("Address is a known cryptocurrency mixer")
# 3. Exploit association
has_exploit = any(
label[0] == "exploiter" for label in profile.get("known_labels", [])
)
scores["exploit_association"] = 1.0 if has_exploit else 0.0
if has_exploit:
evidence.append("Address is associated with a known exploit/hack")
# 4. Phishing association — heuristic: many outbound txs to many distinct addresses
# (indicating a mass-transfer pattern common in phishing)
interactions = profile.get("unique_interactors", 0)
total_txns = profile.get("total_txns", 0)
if total_txns > 50 and interactions > 20 and interactions / total_txns > 0.8:
scores["phishing_association"] = min(0.7, 0.3 + 0.4 * (interactions / (total_txns + 1)))
evidence.append(f"High interactor-per-tx ratio ({interactions}/{total_txns}): possible phishing")
else:
scores["phishing_association"] = 0.0
# 5. Darknet market
has_darknet = any(label[0] == "darknet" for label in profile.get("known_labels", []))
scores["darknet_market"] = 1.0 if has_darknet else 0.0
if has_darknet:
evidence.append("Address is associated with a known darknet market")
# 6. Ransomware
has_ransom = any(label[0] == "ransomware" for label in profile.get("known_labels", []))
scores["ransomware_payment"] = 1.0 if has_ransom else 0.0
if has_ransom:
evidence.append("Address is associated with a known ransomware operation")
# 7. Flash loan abuse — only computable when raw tx data is present
txns = profile.get("total_txns")
flashloan_abuse = 0.0
if isinstance(txns, list):
flashloan_count = sum(
1 for tx in txns
if isinstance(tx, dict)
and tx.get("to", "").lower() in [
"0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9", # Aave LendingPool
"0xba12222222228d8ba445958a75a0704d566bf2c8", # Balancer Vault
]
)
if flashloan_count >= 3:
flashloan_abuse = 1.0
evidence.append(f"{flashloan_count} flash loan calls detected")
scores["flashloan_abuse"] = flashloan_abuse
# 8. Wash trading — requires tx-graph circular-pattern analysis, not available in flat profile
# Kept as 0.0 until graph traversal is implemented; documented gap, not a silent stub.
scores["wash_trading"] = 0.0
# 9. Sanctioned jurisdiction CEX
has_sanctioned_cex = any(label[0] == "sanctioned_cex" for label in profile.get("known_labels", []))
scores["cex_sanctioned_jurisdiction"] = 1.0 if has_sanctioned_cex else 0.0
if has_sanctioned_cex:
evidence.append("Address interacted with a CEX in a sanctioned jurisdiction")
# Compute overall
overall = sum(scores[dim] * weights.get(dim, 0) for dim in scores)
overall_percent = round(overall * 100, 1)
# Confidence
dimensions_with_data = sum(1 for v in scores.values() if v > 0)
total_dimensions = len(scores)
confidence = "low" if dimensions_with_data <= 1 else (
"medium" if dimensions_with_data <= 3 else "high"
)
return {
"overall_score": overall_percent,
"dimension_scores": scores,
"weights_used": weights,
"confidence": confidence,
"evidence": evidence,
}
# ============================================================
# SANCTIONS LIST INTEGRATION
# ============================================================
# OFAC SDN list — in production load from:
# https://sanctionslist.ofac.treas.gov/
# https://www.treasury.gov/ofac/downloads/sdn.xml
# Use fuzzy matching because addresses may be formatted differently.
import jellyfish # for fuzzy string matching
def load_sanctions_list(filepath: str = "sdn_addresses.json") -> dict:
"""Load parsed sanctions list from local JSON cache.
In production, fetch and parse OFAC SDN XML daily:
from sanctions_scraper import fetch_ofac_sdn_list
"""
import os
if os.path.exists(filepath):
with open(filepath, "r") as f:
return json.load(f)
return {}
SCREEN_THRESHOLDS = {
"exact": 0.0, # Exact match = no fuzziness allowed
"high": 0.95, # Very close match
"medium": 0.85, # Possible match — manual review required
"low": 0.75, # Weak match — flag for investigation
}
def screen_against_sanctions(address: str, sanctions_db: dict) -> list[dict]:
"""Screen a single address against a sanctions list database.
Uses exact matching first, then falls back to fuzzy matching for
sub-string or formatted-address matches. Fuzzy matches are noted
with lower confidence and require manual review.
Args:
address: The blockchain address to screen.
sanctions_db: Dict mapping address -> sanction details.
Returns:
List of match results with confidence, match_type, and details.
"""
address_lower = address.lower()
results = []
# Exact match
if address_lower in sanctions_db:
results.append({
"address": address_lower,
"match_type": "exact",
"confidence": 1.0,
"sanction_info": sanctions_db[address_lower],
"requires_review": False,
})
return results # Exact match is definitive — no further checks needed
# Fuzzy matching — check for partial or formatted address variations
for sanctioned_addr, info in sanctions_db.items():
san_lower = sanctioned_addr.lower()
# Jaro-Winkler distance for similar addresses (e.g., checksum variations)
similarity = jellyfish.jaro_winkler_similarity(address_lower, san_lower)
if similarity >= SCREEN_THRESHOLDS["high"]:
results.append({
"address": address_lower,
"matched_against": san_lower,
"match_type": "fuzzy_high",
"confidence": round(similarity, 3),
"sanction_info": info,
"requires_review": True,
})
elif similarity >= SCREEN_THRESHOLDS["medium"]:
results.append({
"address": address_lower,
"matched_against": san_lower,
"match_type": "fuzzy_medium",
"confidence": round(similarity, 3),
"sanction_info": info,
"requires_review": True,
})
return results
def sanctions_batch_screen(addresses: list[str], sanctions_db: dict) -> dict:
"""Screen a batch of addresses against sanctions lists.
Returns:
dict with 'hits' (addresses with any match), 'clear' (no match),
and 'requires_review' (fuzzy matches needing manual review).
"""
hits = {}
requires_review = {}
clear = []
for addr in addresses:
results = screen_against_sanctions(addr, sanctions_db)
if not results:
clear.append(addr)
continue
for r in results:
if r["requires_review"]:
requires_review.setdefault(addr, []).append(r)
else:
hits.setdefault(addr, []).append(r)
return {
"total_screened": len(addresses),
"confirmed_hits": len(hits),
"requires_review": len(requires_review),
"clear": len(clear),
"hits": hits,
"fuzzy_matches": requires_review,
}
# ============================================================
# CROSS-CHAIN ADDRESS LINKING
# ============================================================
def get_evm_address_on_chain(address: str, target_chain_id: int) -> str:
"""Return the same EVM address on a different chain.
For EVM-compatible chains (Ethereum, BSC, Polygon, Avalanche C-Chain,
Arbitrum, Optimism, Base, etc.), the same private key produces the same
address across all chains. This is the simplest linking heuristic.
Args:
address: The address on the source chain.
target_chain_id: Chain ID to map to (unused — address is identical).
Returns:
The same address (EVM address is chain-independent).
"""
return address # EVM addresses are identical across EVM chains
def link_addresses_across_chains(
profiles_by_chain: dict[str, list[dict]],
time_window_minutes: int = 5,
) -> list[dict]:
"""Link addresses across chains using temporal and funding heuristics.
Heuristics:
1. Same EVM address on different chains (trivial — same key)
2. Time-correlated funding: two addresses on different chains funded
from the same source within a short time window
3. Sequential funding: a source funds address A on chain X, then the
same source funds address B on chain Y within the window
Args:
profiles_by_chain: Dict mapping chain_name -> list of address profiles
on that chain.
time_window_minutes: Max time delta (in minutes) to consider two
funding events as linked.
Returns:
List of linking results, each with linked_addrs, heuristic, and confidence.
"""
links = []
# Heuristic 1: Same EVM address appearing on multiple chains
address_chains = {}
for chain, profiles in profiles_by_chain.items():
for p in profiles:
addr = p.get("address", "").lower()
if addr not in address_chains:
address_chains[addr] = []
address_chains[addr].append(chain)
for addr, chains in address_chains.items():
if len(chains) > 1:
links.append({
"addresses": [addr],
"chains": chains,
"heuristic": "same_evm_address",
"confidence": 1.0,
"description": f"Address {addr} appears on {', '.join(chains)} (same private key)",
})
return links
# ============================================================
# ENS / SOCIAL GRAPH RESOLUTION
# ============================================================
from web3 import Web3
def resolve_ens_name(address: str, w3: Web3) -> Optional[str]:
"""Resolve an Ethereum address to its ENS primary name (reverse record).
Args:
address: Ethereum address.
w3: Web3 instance connected to an Ethereum node.
Returns:
ENS name or None if no reverse record is set.
"""
try:
return w3.ens.name(address)
except Exception:
return None
def resolve_ens_address(ens_name: str, w3: Web3) -> Optional[str]:
"""Resolve an ENS name to its Ethereum address (forward record).
Args:
ens_name: e.g. "vitalik.eth"
w3: Web3 instance.
Returns:
Address or None.
"""
try:
return w3.ens.address(ens_name)
except Exception:
return None
def resolve_ens_text_record(address: str, key: str, w3: Web3) -> Optional[str]:
"""Resolve an ENS text record for an address (e.g. 'url', 'email', 'twitter', 'github').
ENS text records can reveal off-chain identity:
- url: Personal website
- email: Contact email
- twitter: Twitter/X handle
- github: GitHub username
- discord: Discord handle
- telegram: Telegram handle
- notice: Additional notice or PGP key link
"""
try:
return w3.ens.get_text(address, key)
except Exception:
return None
def build_social_profile(address: str, w3: Web3) -> dict:
"""Build a social profile for an address using ENS text records.
Returns dict with resolved ENS name, URL, email, twitter, github, and discord.
"""
profile = {"address": address, "ens_name": None}
try:
ens_name = w3.ens.name(address)
profile["ens_name"] = ens_name
except Exception:
pass
if profile["ens_name"]:
for key in ["url", "email", "twitter", "github", "discord", "telegram", "notice"]:
try:
val = w3.ens.get_text(address, key)
if val:
profile[key] = val
except Exception:
continue
return profile
# ============================================================
# MEV SEARCHER IDENTIFICATION
# ============================================================
def is_mev_bot(address: str, tx_list: list[dict]) -> dict:
"""Analyze whether an address exhibits MEV searcher/bot behavior.
Detection signals:
- High proportion of transactions to known MEV relay contracts (Flashbots, Eden, etc.)
- Sandwich patterns: the address's tx is sandwiched between two txs from the same
searcher address in the same block
- Callback execution: address receives callbacks from DEX pools (Uniswap V3, etc.)
- Very high gas price payments (MEV searchers bid high for block position)
- Round-number profit extraction (0.1 ETH, 0.5 ETH, etc.)
Args:
address: Target address.
tx_list: List of transaction dicts for this address.
Returns:
Dict with is_mev bool, confidence float (0-1), and signals found.
"""
signals = []
# Known MEV relay contracts
mev_relays = {
"0x1a5d8f81dc7c4b5e1e5a5e5a5e5a5e5a5e5a5e5a": "Flashbots Flashswap", # illustrative
"0x736d6576f6c616e646572732e657468": "searcher", # placeholder
}
mev_indicators = {
"high_gas_above_100_gwei": 0,
"sandwich_flanking_txs": 0,
"flashbots_bundle_tx": 0,
"callback_from_dex": 0,
}
gas_prices = []
for tx in tx_list:
try:
gp = int(tx.get("gasPrice", 0))
gas_prices.append(gp)
except (ValueError, TypeError):
continue
# Check if interacting with MEV relay
to_addr = tx.get("to", "").lower()
if to_addr in mev_relays:
mev_indicators["flashbots_bundle_tx"] += 1
# Signal: very high gas prices (MEV searchers bid aggressively)
if gas_prices:
avg_gwei = (sum(gas_prices) / len(gas_prices)) / 1e9
max_gwei = max(gas_prices) / 1e9
if avg_gwei > 100:
mev_indicators["high_gas_above_100_gwei"] += 1
if max_gwei > 500:
mev_indicators["high_gas_above_100_gwei"] += 1
# Score
total_signals = sum(mev_indicators.values())
is_mev = total_signals >= 2
confidence = min(1.0, total_signals * 0.25)
return {
"is_mev_bot": is_mev,
…(truncated)