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 = {
"0x0000000000000000000000000000000000000000": "placeholder",
"0x1e227979f6b5c9704f9a92e4201ffc7d7c2d7bbf": "Bybit Exploiter (Lazarus)",
"0x59e0cda5922ef1a80d49f5fe4714e2343dd2ae4f": "Ronin Bridge Exploiter",
"0x098b716b8aaf21512996dc57eb0615e2383e2f96": "Nomad Bridge Exploiter",
"0x0de8f4f3c92abb2fc3a6c4ad07a39bbffa4c37a5": "Wormhole Exploiter",
"0x5dafb0d0f71b5acd3a1d4e21a358e7dcb75bceff": "FTX Drainer",
}
# 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]))
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 (placeholder — requires real list)
scores["darknet_market"] = 0.0
# 6. Ransomware (placeholder — requires real list)
scores["ransomware_payment"] = 0.0
# 7. Flash loan abuse — heuristic: high volume of flash loan calls
flashloan_count = sum(1 for tx in profile.get("total_txns", []) if isinstance(tx, dict) and tx.get("to", "").lower() in [
"0x7d2768de32b0b80b7a3454c06bdac94a69ddc7a9", # Aave LendingPool
"0xba12222222228d8ba445958a75a0704d566bf2c8", # Balancer Vault
])
if isinstance(profile.get("total_txns"), int):
pass # No flash loan detection without tx data in profile
scores["flashloan_abuse"] = 0.0
# 8. Wash trading — heuristic: many txs to self or circular patterns
scores["wash_trading"] = 0.0
# 9. Sanctioned jurisdiction CEX (placeholder)
scores["cex_sanctioned_jurisdiction"] = 0.0
# 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,
"confidence": round(confidence, 2),
"signals": mev_indicators,
"total_signals": total_signals,
"assessment": "MEV searcher/bot" if is_mev else "Likely regular user",
}
# ============================================================
# WALLET AGE / ACTIVITY SCORING
# ============================================================
def compute_wallet_maturity_score(profile: dict) -> dict:
"""Compute a wallet maturity score (0-100) based on age, activity regularity, and protocol diversity.
Factors:
- Wallet age (days since first tx): older = more mature
- Transaction count: more = more active
- Activity regularity: even spacing between transactions indicates organic use
- Protocol diversity: interacting with many distinct protocols = sophisticated
- Number of unique interactors
- Token holding diversity: many token types = diversified
Returns dict with overall maturity score and per-factor breakdown.
"""
score = 0.0
factors = {}
# 1. Age factor (max 25 points)
age_days = profile.get("age_days", 0)
if age_days >= 1095: # 3+ years
age_score = 25
elif age_days >= 365: # 1+ year
age_score = 20
elif age_days >= 90: # 3+ months
age_score = 15
elif age_days >= 30: # 1+ month
age_score = 10
elif age_days >= 7: # 1+ week
age_score = 5
else:
age_score = 0
factors["age"] = age_score
score += age_score
# 2. Activity volume factor (max 25 points)
total_activity = profile.get("total_activity", 0)
if total_activity >= 1000:
activity_score = 25
elif total_activity >= 100:
activity_score = 20
elif total_activity >= 20:
activity_score = 15
elif total_activity >= 5:
activity_score = 10
elif total_activity >= 1:
activity_score = 5
else:
activity_score = 0
factors["activity_volume"] = activity_score
score += activity_score
# 3. Protocol diversity factor (max 25 points)
protocol_count = profile.get("protocols_touched", 0) if isinstance(profile.get("protocols_touched"), int) else len(profile.get("protocols_touched", []))
if protocol_count >= 50:
protocol_score = 25
elif protocol_count >= 20:
protocol_score = 20
elif protocol_count >= 10:
protocol_score = 15
elif protocol_count >= 5:
protocol_score = 10
elif protocol_count >= 1:
protocol_score = 5
else:
protocol_score = 0
factors["protocol_diversity"] = protocol_score
score += protocol_score
# 4. Token holding diversity (max 15 points)
token_count = len(profile.get("token_holdings", []))
if token_count >= 20:
token_score = 15
elif token_count >= 10:
token_score = 12
elif token_count >= 5:
token_score = 8
elif token_count >= 1:
token_score = 4
else:
token_score = 0
factors["token_diversity"] = token_score
score += token_score
# 5. Interaction diversity (max 10 points)
unique_interactors = profile.get("unique_interactors", 0)
if unique_interactors >= 100:
interactor_score = 10
elif unique_interactors >= 50:
interactor_score = 8
elif unique_interactors >= 10:
interactor_score = 5
elif unique_interactors >= 1:
interactor_score = 2
else:
interactor_score = 0
factors["interaction_diversity"] = interactor_score
score += interactor_score
return {
"maturity_score": int(score),
"max_score": 100,
"factors": factors,
"age_days": age_days,
"total_activity": total_activity,
"classification": _classify_maturity(score),
}
def _classify_maturity(score: int) -> str:
"""Classify a maturity score into a descriptive bucket."""
if score >= 80:
return "Very mature — long-established, diversified, highly active wallet"
if score >= 60:
return "Mature — established wallet with solid activity and diversity"
if score >= 40:
return "Developing — moderate activity, some diversity"
if score >= 20:
return "New but active — young wallet with growing usage"
if score >= 1:
return "Nascent — very recent or low-activity wallet"
return "Inactive or zero-activity wallet"
# ============================================================
# FULL INTELLIGENCE REPORT
# ============================================================
def generate_intelligence_report(address: str, etherscan_key: str, w3: Optional[Web3] = None) -> dict:
"""Generate a comprehensive address intelligence report.
This is the main orchestration function that runs all analysis modules
and returns a structured report.
"""
# Step 1: Build base profile
profile = build_address_profile(address, etherscan_key)
# Step 2: Compute maturity score
maturity = compute_wallet_maturity_score(profile)
# Step 3: Risk scoring
risk = compute_risk_score(profile, address)
# Step 4: MEV analysis (requires tx list)
tx_list = _fetch_tx_list(address, etherscan_key)
mev = is_mev_bot(address, tx_list)
# Step 5: ENS resolution (requires web3 connection)
social = {"ens_name": None}
if w3 is not None:
social = build_social_profile(address, w3)
# Step 6: Cross-chain (placeholder — requires multi-chain profiles)
cross_chain = {"note": "Provide profiles_by_chain for cross-chain linking"}
return {
"intelligence_report": {
"address": address,
"generated_at": datetime.now(timezone.utc).isoformat(),
"executive_summary": (
f"Address {address[:10]}...{address[-6:]} "
f"is a {maturity['classification']}. "
f"Risk score: {risk['overall_score']}/100 ({risk['confidence']} confidence). "
f"MEV assessment: {mev['assessment']}. "
f"ENS: {social.get('ens_name', 'Not set')}."
),
},
"profile": {
"eth_balance": profile.get("eth_balance", 0),
"total_activity": profile.get("total_activity", 0),
"first_seen": str(profile.get("first_seen", "N/A")),
"last_active": str(profile.get("last_active", "N/A")),
"age_days": profile.get("age_days", 0),
"unique_interactors": profile.get("unique_interactors", 0),
"gas_profile": profile.get("gas_profile", {}),
"token_holdings": profile.get("token_holdings", []),
},
"maturity": maturity,
"risk_assessment": risk,
"mev_analysis": mev,
"social_profile": social,
"cross_chain": cross_chain,
}
