Extracting Blockchain Data

When analyzing blockchain activity, the first step is accessing historical blockchain data. One common method is using blockchain node APIs, which allow applications to request information directly from a node. Many networks provide public API endpoints that return structured data, often in JSON format, making it easier to process and analyze.

Another approach is running your own blockchain node and querying it through an RPC (Remote Procedure Call) interface. This provides full access to blockchain history and greater control over data retrieval, although it requires more setup and storage. Some projects also provide third-party data services or data dumps, but API and node queries remain the most flexible methods for blockchain data analysis.

from web3 import Web3

# Connect to a local or remote Ethereum node
w3 = Web3(Web3.HTTPProvider("http://localhost:8545"))

start_block = 1000000
end_block = 1000005
all_transactions = []

for block_number in range(start_block, end_block + 1):
    block = w3.eth.get_block(block_number, full_transactions=True)
    for tx in block.transactions:
        tx_data = {
            "block_number": block_number,
            "hash": tx.hash.hex(),
            "from": tx["from"],
            "to": tx["to"],
            "value": tx["value"]
        }
        all_transactions.append(tx_data)

print(f"Extracted {len(all_transactions)} transactions from blocks {start_block} to {end_block}.")

Extracting blockchain data usually involves iterating through a range of blocks and collecting transaction details from each one. In the code above, the program retrieves each block, accesses its transactions, and loops through them.

For every transaction, it extracts fields such as the transaction hash, sender, recipient, and value, storing them in a list of dictionaries. This structure makes the data easier to analyze or visualize later.

# Transactions sent to a specific address
target_address = "0x1234567890abcdef1234567890abcdef12345678"
filtered_txs = [
    tx for tx in all_transactions
    if tx["to"] == target_address
]

print(f"{len(filtered_txs)} txs to {target_address}")

# Transactions greater than 1 ETH
high_value_txs = [
    tx for tx in all_transactions
    if tx["value"] > 1_000_000_000_000_000_000
]

print(f"{len(high_value_txs)} txs > 1 ETH")