Output creation by address type over the last 30 days

00_output.txt

Connected to Bitcoin node. Current block height: 905108
Analyzing last 12960 blocks...
Starting from block: 00000000000000000000e9900f08f6dc109762d09d4d0b31b8ece5ab35a5401e
Processing block 12960/12960...
Processed 12960 blocks

============================================================
Bitcoin Address Type Analysis
============================================================
Total Value Analyzed: 53757862.95514309 BTC
============================================================
Address Type    Value (BTC)     Percentage
--------------- --------------- ------------
P2WPKH          24669875.36007922 45.89       %
P2PKH           10933422.73360267 20.34       %
P2WSH           9675588.09990511 18.00       %
P2SH            4998648.88591792 9.30        %
P2TR            3480305.00136064 6.47        %
NONSTANDARD     21.46303677     0.00        %
MULTISIG        1.33157272      0.00        %
OP_RETURN       0.07332491      0.00        %
WITNESS_UNKNOWN 0.00500883      0.00        %
P2PK            0.00102052      0.00        %
UNKNOWN         0.00031378      0.00        %
------------------------------------------
TOTAL           53757862.95514309 100.00%

01_addresstally.py

#!/usr/bin/env -S uv run
# /// script
# dependencies = [
#     "python-bitcoinrpc>=1.0",
# ]
# ///
"""
Bitcoin Address Type Analysis Script

This script analyzes the last n blocks to provide a percentage breakdown
of transaction outputs by address type (P2TR, P2WSH, P2WPKH, etc.) by value.

Requirements:
- Bitcoin Core node running with RPC enabled
- uv installed (https://docs.astral.sh/uv/)

Usage:
    uv run bitcoin_address_analysis.py <rpc_user> <rpc_password> <num_blocks> [rpc_host] [rpc_port]
    
Example:
    uv run bitcoin_address_analysis.py myuser mypass 100
"""

import json
import sys
from decimal import Decimal
from collections import defaultdict
from bitcoinrpc.authproxy import AuthServiceProxy, JSONRPCException

class BitcoinAddressAnalyzer:
    def __init__(self, rpc_user, rpc_password, rpc_host="127.0.0.1", rpc_port=8332):
        """Initialize Bitcoin RPC connection"""
        self.rpc_connection = AuthServiceProxy(
            f"http://{rpc_user}:{rpc_password}@{rpc_host}:{rpc_port}/"
        )
        
    def get_address_type_from_output(self, output):
        """
        Determine address type from transaction output
        Returns the script type based on scriptPubKey
        """
        script_type = output.get('scriptPubKey', {}).get('type', 'unknown')
        
        # Map Bitcoin Core script types to common address types
        type_mapping = {
            'pubkeyhash': 'P2PKH',           # Legacy
            'scripthash': 'P2SH',            # Script Hash (could be wrapped segwit)
            'witness_v0_keyhash': 'P2WPKH',  # Native SegWit v0
            'witness_v0_scripthash': 'P2WSH', # Native SegWit v0 Script
            'witness_v1_taproot': 'P2TR',    # Taproot
            'pubkey': 'P2PK',                # Pay to Public Key (rare)
            'nulldata': 'OP_RETURN',         # Data outputs
            'multisig': 'MULTISIG',          # Bare multisig
            'nonstandard': 'NONSTANDARD',    # Non-standard scripts
            'witness_unknown': 'WITNESS_UNKNOWN'  # Unknown witness versions
        }
        
        return type_mapping.get(script_type, 'UNKNOWN')
    
    def analyze_block(self, block_hash):
        """
        Analyze a single block and return address type breakdown
        """
        try:
            # Get block data with transactions
            block = self.rpc_connection.getblock(block_hash, 2)
            
            type_values = defaultdict(Decimal)
            total_value = Decimal('0')
            
            # Iterate through all transactions in the block
            for tx in block['tx']:
                # Skip coinbase transaction outputs (newly minted coins)
                if tx.get('vin', [{}])[0].get('coinbase'):
                    continue
                
                # Analyze each output
                for output in tx.get('vout', []):
                    value = Decimal(str(output['value']))
                    addr_type = self.get_address_type_from_output(output)
                    
                    type_values[addr_type] += value
                    total_value += value
            
            return type_values, total_value
            
        except JSONRPCException as e:
            print(f"RPC Error analyzing block {block_hash}: {e}")
            return defaultdict(Decimal), Decimal('0')
    
    def analyze_recent_blocks(self, num_blocks):
        """
        Analyze the last n blocks for address type breakdown
        """
        try:
            # Get current best block hash
            best_block_hash = self.rpc_connection.getbestblockhash()
            current_hash = best_block_hash
            
            print(f"Analyzing last {num_blocks} blocks...")
            print(f"Starting from block: {current_hash}")
            
            total_type_values = defaultdict(Decimal)
            grand_total = Decimal('0')
            blocks_processed = 0
            
            # Walk backwards through the blockchain
            for i in range(num_blocks):
                if not current_hash:
                    break
                
                print(f"Processing block {i+1}/{num_blocks}...", end='\r')
                
                type_values, block_total = self.analyze_block(current_hash)
                
                # Accumulate totals
                for addr_type, value in type_values.items():
                    total_type_values[addr_type] += value
                grand_total += block_total
                blocks_processed += 1
                
                # Get previous block hash
                try:
                    block_info = self.rpc_connection.getblock(current_hash, 1)
                    current_hash = block_info.get('previousblockhash')
                except JSONRPCException:
                    break
            
            print(f"\nProcessed {blocks_processed} blocks")
            
            return total_type_values, grand_total
            
        except JSONRPCException as e:
            print(f"RPC Error: {e}")
            return defaultdict(Decimal), Decimal('0')
    
    def print_analysis(self, type_values, total_value):
        """
        Print formatted analysis results
        """
        if total_value == 0:
            print("No transaction outputs found to analyze.")
            return
        
        print(f"\n{'='*60}")
        print(f"Bitcoin Address Type Analysis")
        print(f"{'='*60}")
        print(f"Total Value Analyzed: {total_value:.8f} BTC")
        print(f"{'='*60}")
        
        # Sort by value (descending)
        sorted_types = sorted(type_values.items(), key=lambda x: x[1], reverse=True)
        
        print(f"{'Address Type':<15} {'Value (BTC)':<15} {'Percentage':<12}")
        print(f"{'-'*15} {'-'*15} {'-'*12}")
        
        for addr_type, value in sorted_types:
            percentage = (value / total_value * 100) if total_value > 0 else 0
            print(f"{addr_type:<15} {value:<15.8f} {percentage:<12.2f}%")
        
        print(f"{'-'*42}")
        print(f"{'TOTAL':<15} {total_value:<15.8f} {'100.00%':<12}")

def main():
    """
    Main function to run the analysis
    """
    if len(sys.argv) < 4:
        print("Usage: uv run bitcoin_address_analysis.py <rpc_user> <rpc_password> <num_blocks> [rpc_host] [rpc_port]")
        print("Example: uv run bitcoin_address_analysis.py myuser mypass 100")
        sys.exit(1)
    
    rpc_user = sys.argv[1]
    rpc_password = sys.argv[2]
    num_blocks = int(sys.argv[3])
    rpc_host = sys.argv[4] if len(sys.argv) > 4 else "127.0.0.1"
    rpc_port = int(sys.argv[5]) if len(sys.argv) > 5 else 8332
    
    try:
        analyzer = BitcoinAddressAnalyzer(rpc_user, rpc_password, rpc_host, rpc_port)
        
        # Test connection
        try:
            block_count = analyzer.rpc_connection.getblockcount()
            print(f"Connected to Bitcoin node. Current block height: {block_count}")
        except JSONRPCException as e:
            print(f"Failed to connect to Bitcoin node: {e}")
            sys.exit(1)
        
        # Run analysis
        type_values, total_value = analyzer.analyze_recent_blocks(num_blocks)
        
        # Print results
        analyzer.print_analysis(type_values, total_value)
        
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()