{"w3":{"description":"Web3 instance for blockchain connection","required":true,"example":"Web3(Web3.HTTPProvider('https://ethereum-rpc.publicnode.com'))"},"addresses":{"description":"List of addresses to analyze","required":true,"example":"['0xabc...', '0xdef...']"},"transactions":{"description":"Transaction data for graph analysis","required":true},"start_block":{"description":"Start block for analysis window","required":false},"end_block":{"description":"End block for analysis window","required":false}}
Advanced detection system for identifying Sybil attacks, bot networks, and insider trading patterns using multi-heuristic analysis, machine learning clustering, and transaction graph algorithms on real blockchain data.
Cluster buys within time window (default 5 minutes)
Flag clusters with >= 3 addresses
Calculate confidence based on cluster size
Evidence: Cluster size, time window, actual time span, total volume
Detection Engine Pipeline
Unified 4-Phase Architecture
Phase 1: Address Clustering
Apply common input heuristic
Detect change addresses
Analyze funding patterns
Perform temporal correlation
Phase 2: Graph Analysis
Build transaction graph
Detect communities
Find star patterns
Detect chain patterns
Phase 3: Behavior Profiling
Profile each address
Detect anomalies
Identify bot patterns
Phase 4: Insider Detection
Scan for token launches
Detect pre-launch accumulation
Check pre-announcement activity
Find coordinated buying
Threat Level Classification
classThreatLevel(Enum):
LOW = "low"# Minimal risk
MEDIUM = "medium"# Suspicious but not confirmed
HIGH = "high"# Strong evidence
CRITICAL = "critical"# Confirmed threat
Multi-Factor Scoring
Combines evidence from multiple detection methods with confidence calculation based on address cluster size, timing precision, and volume concentration.
Usage Examples
Example 1: Basic Address Analysis
from web3 import Web3
from address_clustering import AddressClustering
import os
# Connect to Ethereum
w3 = Web3(Web3.HTTPProvider(os.getenv("RPC_URL")))
clusterer = AddressClustering(w3)
# Analyze addresses
addresses = [
'0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045', # vitalik.eth'0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48', # USDC'0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'# WETH
]
# Extract features
features = [
clusterer.extract_address_features(addr, 18000000, 18100000)
for addr in addresses
]
# Perform clustering
clusters = clusterer.cluster_by_behavior(
features,
algorithm='kmeans',
n_clusters=3
)
# Print resultsfor cluster in clusters:
print(f"Cluster {cluster.cluster_id}: {cluster.cluster_type}")
print(f" Confidence: {cluster.confidence:.2%}")
print(f" Addresses: {len(cluster.addresses)}")
Expected Output:
INFO:address_clustering:Extracting features for address...
INFO:address_clustering:Clustering 3 address features
INFO:address_clustering:K-Means clustering complete
Cluster 0: normal
Confidence: 75.23%
Addresses: 2
Cluster 1: normal
Confidence: 68.45%
Addresses: 1
Example 2: Transaction Graph Analysis
from graph_analyzer import GraphAnalyzer
analyzer = GraphAnalyzer(w3)
# Build graph from transactions
transactions = [
{'from': '0xa...', 'to': '0xb...', 'value': 1.5, 'timestamp': 1000},
{'from': '0xa...', 'to': '0xc...', 'value': 1.5, 'timestamp': 1001},
# ... more transactions
]
graph = analyzer.build_transaction_graph(transactions)
# Detect communities
communities = analyzer.detect_communities(algorithm='louvain')
print(f"Found {len(communities)} communities")
for comm_id, addresses in communities.items():
print(f" Community {comm_id}: {len(addresses)} addresses")
# Find star patterns
stars = analyzer.detect_star_patterns(min_connections=5)
for network in stars:
print(f"\n🚨 Star network detected!")
print(f" Hub: {network.hub_address}")
print(f" Connected: {len(network.addresses)} addresses")
print(f" Total volume: {network.total_volume:.2f} ETH")
print(f" Confidence: {network.confidence:.2%}")
from behavior_profiler import BehaviorProfiler
profiler = BehaviorProfiler(w3)
# Profile address
address = '0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb'
addr_transactions = [...] # Get transaction history
profile = profiler.profile_address(address, addr_transactions)
print(f"Profile for {profile.address}")
print(f"\nTemporal Patterns:")
print(f" Regularity score: {profile.activity_regularity_score:.2f}")
print(f" Avg tx/day: {profile.avg_tx_per_day:.1f}")
print(f"\nGas Behavior:")
print(f" Avg gas price: {profile.avg_gas_price:.1f} Gwei")
print(f" Optimization score: {profile.gas_optimization_score:.2f}")
print(f"\nAnomaly Detection:")
print(f" Is anomalous: {profile.is_anomalous}")
print(f" Anomaly score: {profile.anomaly_score:.2f}")
if profile.anomaly_reasons:
print(f" Reasons:")
for reason in profile.anomaly_reasons:
print(f" - {reason}")
Test Output (from test execution):
INFO:behavior_profiler:Profiling address: 0xa
✅ Profile created for: 0xa
Temporal Patterns:
Activity regularity: 0.50
Avg tx/day: 12960.0
Gas Behavior:
Avg gas price: 50.0 Gwei
Gas variance: 0.00
Gas optimization score: 0.90
Value Patterns:
Avg value: 1.0000 ETH
Median value: 1.0000 ETH
Anomaly Detection:
Anomaly score: 0.80
Is anomalous: True
Reasons:
- Consistent gas pricing (bot-like)
- Identical transaction values
- Extremely high transaction frequency
- Activity concentrated in single hour
Example 4: Insider Trading Detection
from insider_detector import InsiderDetector
detector = InsiderDetector(w3)
# Detect pre-launch accumulation
token_address = '0xTokenContractAddress'
launch_block = 18500000
event = detector.detect_pre_launch_accumulation(
token_address,
launch_block,
lookback_blocks=1000,
min_addresses=3
)
if event:
print(f"🚨 INSIDER TRADING DETECTED!")
print(f" Type: {event.event_type}")
print(f" Token: {event.token_address}")
print(f" Confidence: {event.confidence:.2%}")
print(f" Addresses involved: {len(event.addresses)}")
print(f" Total volume: {event.total_volume:.2f}")
print(f"\n Evidence:")
for key, value in event.evidence.items():
print(f" {key}: {value}")
else:
print("✅ No insider trading detected")
Test Output (from test execution):
INFO:insider_detector:Detecting pre-launch accumulation for 0x1234567890...
INFO:insider_detector:No pre-launch activity found
✅ No insider trading detected (expected - no real activity)
Example 5: Comprehensive Detection (Main Engine)
from detector_engine import DetectorEngine
# Initialize engine with custom thresholds
engine = DetectorEngine(
w3,
sybil_threshold=0.6,
insider_threshold=0.7,
bot_threshold=0.5
)
# Run comprehensive analysis
addresses = [...] # List of addresses to analyze
transactions = [...] # Transaction data
report = engine.analyze_addresses(
addresses,
transactions,
start_block=18000000,
end_block=18100000
)
# Print summaryprint(f"Analysis Report: {report.report_id}")
print(f"Addresses analyzed: {report.total_addresses_analyzed}")
print(f"Transactions analyzed: {report.total_transactions_analyzed}")
print(f"\nThreats detected: {report.total_threats}")
print(f" Critical: {report.critical_threats}")
print(f" High: {report.high_threats}")
# Export as JSON
json_report = engine.export_report(report, format='json')
print(json_report)
# Export as text
text_report = engine.export_report(report, format='text')
print(text_report)
