Blockchain Analysis

Tools for analyzing blockchain transactions and cryptocurrency data.

Transaction Parsing

import hashlib
import json

class Transaction:
    def __init__(self, sender, receiver, amount, timestamp):
        self.sender = sender
        self.receiver = receiver
        self.amount = amount
        self.timestamp = timestamp
        self.hash = self.calculate_hash()

    def calculate_hash(self):
        tx_string = f"{self.sender}{self.receiver}{self.amount}{self.timestamp}"
        return hashlib.sha256(tx_string.encode()).hexdigest()

    def to_dict(self):
        return {
            'sender': self.sender,
            'receiver': self.receiver,
            'amount': self.amount,
            'timestamp': self.timestamp,
            'hash': self.hash
        }

Address Clustering

import networkx as nx

def cluster_addresses(transactions):
    """
    Cluster addresses based on common input heuristic.
    Addresses used as inputs in same transaction are likely same entity.
    """
    G = nx.Graph()

    for tx in transactions:
        # Add edges between co-spent inputs
        inputs = tx.get('inputs', [])
        for i, addr1 in enumerate(inputs):
            for addr2 in inputs[i+1:]:
                G.add_edge(addr1, addr2)

    # Find connected components (clusters)
    clusters = list(nx.connected_components(G))
    return clusters

def label_clusters(clusters, known_entities):
    """Label clusters with known entity names."""
    labeled = {}
    for i, cluster in enumerate(clusters):
        for addr in cluster:
            if addr in known_entities:
                labeled[i] = known_entities[addr]
                break
        else:
            labeled[i] = f"Unknown_{i}"
    return labeled

Flow Analysis

import pandas as pd

def calculate_flow_metrics(transactions_df):
    """Calculate flow metrics for addresses."""

    # Aggregate inflows and outflows
    inflows = transactions_df.groupby('receiver')['amount'].sum()
    outflows = transactions_df.groupby('sender')['amount'].sum()

    # Net flow
    all_addresses = set(inflows.index) | set(outflows.index)
    metrics = []

    for addr in all_addresses:
        total_in = inflows.get(addr, 0)
        total_out = outflows.get(addr, 0)
        metrics.append({
            'address': addr,
            'total_received': total_in,
            'total_sent': total_out,
            'net_flow': total_in - total_out,
            'tx_count': len(transactions_df[
                (transactions_df['sender'] == addr) |
                (transactions_df['receiver'] == addr)
            ])
        })

    return pd.DataFrame(metrics)

Taint Analysis

def trace_taint(transactions, source_address, max_hops=5):
    """
    Trace funds from a source address through the network.
    """
    tainted = {source_address: 1.0}  # address -> taint level
    visited = set()

    for hop in range(max_hops):
        new_tainted = {}
        for tx in transactions:
            if tx['sender'] in tainted and tx['sender'] not in visited:
                # Propagate taint to receiver
                receiver = tx['receiver']
                if receiver not in new_tainted:
                    new_tainted[receiver] = 0
                new_tainted[receiver] += tainted[tx['sender']] * 0.5

        visited.update(tainted.keys())
        tainted.update(new_tainted)

    return tainted

Web3 Integration

# Web3.py for Ethereum interaction
from web3 import Web3

def get_transaction_details(tx_hash, provider_url):
    """Get transaction details from Ethereum."""
    w3 = Web3(Web3.HTTPProvider(provider_url))

    tx = w3.eth.get_transaction(tx_hash)
    receipt = w3.eth.get_transaction_receipt(tx_hash)

    return {
        'from': tx['from'],
        'to': tx['to'],
        'value': w3.from_wei(tx['value'], 'ether'),
        'gas_used': receipt['gasUsed'],
        'status': receipt['status']
    }