Building a Candidate Block Builder in Pure Python

To understand how a miner's template engine constructs candidate blocks, we can implement a custom Block Builder in Python.

By writing this from scratch without external libraries, we can model how unconfirmed transactions are prioritized, how multidimensional weight and SigOp caps are enforced, and how a valid Merkle root is recursively calculated to generate the final 80-byte block header.

💻 1. The Block Builder Simulator Source Code

The following standard-library Python script implements a complete block builder: 1. Transaction Modeling: Models transactions with virtual sizes, fees, parent dependencies, and signature operations (SigOps). 2. Greedy Packing Algorithm: Implements a multidimensional knapsack-aware package selection engine. 3. Recursive Merkle Tree Engine: Calculates the 32-byte Merkle Root, handling odd transaction counts. 4. Header Serializer: Serializes and outputs the resulting mock Block Header.

import hashlib
import struct
import time
from typing import List, Dict, Set

class SimulatedTx:
    def __init__(self, txid: str, size: int, fee: int, sigops: int, parents: List[str] = None):
        self.txid = txid
        self.size = size      # Size in vBytes
        self.fee = fee        # Fee in Satoshis
        self.sigops = sigops  # Signature operations
        self.parents = set(parents) if parents else set()

class BlockBuilder:
    def __init__(self, max_weight_kb: float, max_sigops: int):
        self.max_weight_kb = max_weight_kb
        self.max_sigops = max_sigops
        self.mempool: Dict[str, SimulatedTx] = {}

    def add_to_mempool(self, tx: SimulatedTx):
        self.mempool[tx.txid] = tx

    def calculate_double_sha256(self, data: bytes) -> bytes:
        """Returns the double-SHA256 hash of a byte array."""
        return hashlib.sha256(hashlib.sha256(data).digest()).digest()

    def calculate_merkle_root(self, txids: List[str]) -> str:
        """Recursively calculates the 32-byte Merkle Root from transaction IDs."""
        # Convert hex TXID strings to raw 32-byte arrays in little-endian
        current_level = [bytes.fromhex(txid)[::-1] for txid in txids]

        if not current_level:
            return "00" * 32

        while len(current_level) > 1:
            next_level = []
            for i in range(0, len(current_level), 2):
                left = current_level[i]
                # If there is no right partner, duplicate the left child (odd leaf rule)
                right = current_level[i+1] if (i + 1) < len(current_level) else left
                parent = self.calculate_double_sha256(left + right)
                next_level.append(parent)
            current_level = next_level

        # Return hex-encoded Merkle Root in big-endian representation
        return current_level[0][::-1].hex()

    def assemble_candidate_block(self) -> List[str]:
        """Assembles a candidate block, respecting limits and sorting packages."""
        print("[*] Initiating Candidate Block Assembly...")

        # 1. Initialize the block with a Coinbase Transaction (always leaf 0)
        coinbase_txid = "a05c1d428fa69399db6a65502c31e626db4bf6ef4f1d4f2bd3f45f2cd3f45f2c"
        selected_txids = [coinbase_txid]

        current_weight_kb = 0.5  # Reserve 0.5 KB for block overhead and Coinbase
        current_sigops = 1       # Coinbase has 1 SigOp
        total_fees_collected = 0

        # 2. Sort unconfirmed mempool transactions by individual feerate (simplifying package sorting)
        sorted_mempool = sorted(self.mempool.values(), key=lambda tx: tx.fee / tx.size, reverse=True)

        # 3. Greedy Knapsack packing loop
        for tx in sorted_mempool:
            # Check if including this transaction would violate block boundaries
            tx_weight_kb = tx.size / 1024.0

            if (current_weight_kb + tx_weight_kb) > self.max_weight_kb:
                print(f"[!] Skipped {tx.txid}: Exceeds weight budget limit.")
                continue

            if (current_sigops + tx.sigops) > self.max_sigops:
                print(f"[!] Skipped {tx.txid}: Exceeds SigOps CPU limit.")
                continue

            # Verify unconfirmed parent dependencies are already packed ahead of it
            parents_missing = [p for p in tx.parents if p not in selected_txids]
            if parents_missing:
                print(f"[!] Skipped {tx.txid}: Missing parent dependencies ({parents_missing}).")
                continue

            # Include transaction in block
            selected_txids.append(tx.txid)
            current_weight_kb += tx_weight_kb
            current_sigops += tx.sigops
            total_fees_collected += tx.fee
            print(f"[+] Packed Tx {tx.txid} (Size: {tx.size} vB | Fee: {tx.fee} Sats | SigOps: {tx.sigops})")

        print("\n=== Candidate Block Statistics ===")
        print(f"Total Transactions Packed: {len(selected_txids)}")
        print(f"Block Weight: {current_weight_kb:.2f} KB / {self.max_weight_kb:.2f} KB")
        print(f"Block SigOps: {current_sigops} / {self.max_sigops}")
        print(f"Transaction Fees Collected: {total_fees_collected} Sats")
        return selected_txids

    def serialize_block_header(self, merkle_root: str) -> bytes:
        """Serializes standard 80-byte block header format."""
        version = 536870912  # Version 4 block (BIP 9 active)
        prev_block_hash = "0000000000000000000201a4e1564f26db9b6426db4bf6ef4f1d4f2bd3f45f2c"
        timestamp = int(time.time())
        bits = 0x1703a3d5      # Compact difficulty target
        nonce = 18429381       # Dummy Nonce

        # Convert hex strings to raw byte arrays (LE representation)
        prev_block_bytes = bytes.fromhex(prev_block_hash)[::-1]
        merkle_root_bytes = bytes.fromhex(merkle_root)[::-1]

        # Structure format: <i (version) + 32s (prev hash) + 32s (merkle root) + I (time) + I (bits) + I (nonce)
        header_packed = struct.pack(
            "<i32s32sIII",
            version,
            prev_block_bytes,
            merkle_root_bytes,
            timestamp,
            bits,
            nonce
        )
        return header_packed

if __name__ == "__main__":
    # Create block builder with highly constrained test parameters (1.5 KB max weight, 10 max SigOps)
    builder = BlockBuilder(max_weight_kb=1.5, max_sigops=10)

    # Ingest unconfirmed transactions to local Mempool
    # tx1: Standard transaction
    builder.add_to_mempool(SimulatedTx("1111111111111111111111111111111111111111111111111111111111111111", size=400, fee=4000, sigops=2))
    # tx2: High SigOp transaction
    builder.add_to_mempool(SimulatedTx("2222222222222222222222222222222222222222222222222222222222222222", size=300, fee=6000, sigops=9))
    # tx3: Low-fee standard transaction
    builder.add_to_mempool(SimulatedTx("3333333333333333333333333333333333333333333333333333333333333333", size=300, fee=900, sigops=1))
    # tx4: High-fee child dependent on parent tx1
    builder.add_to_mempool(SimulatedTx("4444444444444444444444444444444444444444444444444444444444444444", size=200, fee=5000, sigops=1, parents=["1111111111111111111111111111111111111111111111111111111111111111"]))

    # 1. Assemble the candidate block
    selected_txs = builder.assemble_candidate_block()

    # 2. Compute the cryptographic Merkle Root
    merkle_root = builder.calculate_merkle_root(selected_txs)
    print(f"\n[✔] Calculated Cryptographic Merkle Root:\n    {merkle_root}")

    # 3. Serialize block header
    header_bytes = builder.serialize_block_header(merkle_root)
    print(f"\n[✔] Serialized 80-Byte Candidate Block Header (Hex):\n    {header_bytes.hex()}")