Building a Blockchain Height and Reorg Simulator in Pure Python

To understand how Bitcoin nodes track block height, evaluate cumulative cryptographic work (nChainWork), and execute Blockchain Reorganizations (Reorgs) when competing chain tips emerge, we can implement a custom Blockchain Reorg Simulator in Python.

By writing this simulator from scratch using only Python's standard libraries, we can model how nodes re-route their active chain tip to follow the heaviest proof-of-work path.

💻 1. The Blockchain Reorg Simulator Source Code

Save the following standard-library Python script as python_reorg_simulator.py.

import hashlib
import json

class Block:
    def __init__(self, block_id: str, prev_id: str, height: int, difficulty: int, transactions: list):
        self.block_id = block_id
        self.prev_id = prev_id
        self.height = height
        # Difficulty represents the work weight of this block
        self.difficulty = difficulty
        self.transactions = transactions
        # Cumulative work on this branch
        self.cumulative_work = difficulty

    def to_dict(self):
        return {
            "block_id": self.block_id,
            "prev_id": self.prev_id,
            "height": self.height,
            "difficulty": self.difficulty,
            "cumulative_work": self.cumulative_work,
            "transactions": self.transactions
        }

class BlockchainSimulator:
    def __init__(self):
        # Master block database storing blocks index strictly by block_id
        self.block_index = {}
        # Track active chain tip
        self.active_tip_id = "0"

        # Create and initialize Genesis Block (Height 0)
        genesis = Block(
            block_id="GENESIS_BLOCK_0",
            prev_id="00000000000000000000000000000000",
            height=0,
            difficulty=1000, # Hardcoded starting work weight
            transactions=["Coinbase: Pay Satoshi 50 BTC"]
        )
        genesis.cumulative_work = 1000
        self.block_index[genesis.block_id] = genesis
        self.active_tip_id = genesis.block_id

    def get_block(self, block_id: str) -> Block:
        return self.block_index.get(block_id)

    def add_block(self, block_id: str, prev_id: str, difficulty: int, transactions: list) -> str:
        """
        Mines and attempts to attach a new block to the blockchain.
        Evaluates whether this block triggers a chain reorganization (reorg).
        """
        prev_block = self.get_block(prev_id)
        if not prev_block:
            raise ValueError(f"Parent block {prev_id} does not exist in block index database.")

        height = prev_block.height + 1
        new_block = Block(block_id, prev_id, height, difficulty, transactions)

        # Calculate new cumulative work on this branch
        new_block.cumulative_work = prev_block.cumulative_work + difficulty
        self.block_index[block_id] = new_block

        # Check if we need to reorganize the active chain tip
        active_tip = self.get_block(self.active_tip_id)

        print(f"\n[*] Mined Block '{block_id}' at Height {height} (Work: {difficulty}, Cumulative: {new_block.cumulative_work})")

        if new_block.cumulative_work > active_tip.cumulative_work:
            # If the new block represents more total work, swap active tips!
            if new_block.prev_id == self.active_tip_id:
                # Simple tip extension (no reorganization needed)
                self.active_tip_id = block_id
                print(f"    ├─► Active tip successfully extended to '{block_id}'")
            else:
                # Trigger a chain reorganization!
                print(f"    ⚠️⚠️⚠️ CHAIN REORGANIZATION DETECTED! ⚠️⚠️⚠️")
                self._execute_reorg(old_tip_id=self.active_tip_id, new_tip_id=block_id)
        else:
            print(f"    ├─► Block attached to fork branch. Remains an ORPHAN branch (Cumulative Work is lower: {new_block.cumulative_work} vs Active {active_tip.cumulative_work})")

        return self.active_tip_id

    def _execute_reorg(self, old_tip_id: str, new_tip_id: str):
        """
        Traces back branches to find the common ancestor, rolls back transactions, and applies new branch.
        """
        old_branch = []
        new_branch = []

        curr_old = old_tip_id
        curr_new = new_tip_id

        # Step 1: Trace back the branches to locate the fork intersection block
        while curr_old != curr_new:
            old_block = self.get_block(curr_old)
            new_block = self.get_block(curr_new)

            if old_block.height >= new_block.height:
                old_branch.append(old_block)
                curr_old = old_block.prev_id
            if new_block.height > old_block.height:
                new_branch.append(new_block)
                curr_new = new_block.prev_id

        common_ancestor_id = curr_old
        print(f"    ├─► Common Ancestor Block located: '{common_ancestor_id}'")

        # Step 2: Rollback (disconnect) old branch blocks
        print("    ├─► ROLLBACK IN PROGRESS:")
        for block in old_branch:
            print(f"    │   └─ Disconnected Block '{block.block_id}' from height {block.height}. Returned txs {block.transactions} to Mempool.")

        # Step 3: Connect new branch blocks
        print("    ├─► CONNECTING NEW BRANCH:")
        for block in reversed(new_branch):
            print(f"    │   └─ Applied Block '{block.block_id}' to height {block.height}. Executed txs {block.transactions}.")

        # Update active tip
        self.active_tip_id = new_tip_id
        print(f"    [✔] Reorg Complete! Active tip is now '{self.active_tip_id}' (Height: {self.get_block(self.active_tip_id).height})")

if __name__ == "__main__":
    simulator = BlockchainSimulator()

    print("=== Initializing Blockchain Reorg Simulator ===")
    print(f"[*] Genesis block is set: {simulator.active_tip_id}")

    # 1. Build a normal linear chain (Height 1 and 2)
    simulator.add_block("BLOCK_1", "GENESIS_BLOCK_0", difficulty=100, transactions=["Tx: Alice to Bob 1 BTC"])
    simulator.add_block("BLOCK_2", "BLOCK_1", difficulty=100, transactions=["Tx: Bob to Charlie 0.5 BTC"])

    # 2. Mine a competing Block 3A (difficulty 100)
    simulator.add_block("BLOCK_3A", "BLOCK_2", difficulty=100, transactions=["Tx: Charlie to David 0.2 BTC"])

    # 3. Create a Fork! An attacker mines a competing Block 3B from Block 2 (with HIGHER difficulty, say 150)
    # This represents a heavier block mined with more electrical work.
    simulator.add_block("BLOCK_3B", "BLOCK_2", difficulty=150, transactions=["Tx: Double Spend Attempt! Charlie to Eve 0.5 BTC"])

    # 4. Extend the weaker branch to Block 4A
    simulator.add_block("BLOCK_4A", "BLOCK_3A", difficulty=20, transactions=["Tx: David to Frank 0.1 BTC"])

    # 5. Extend the stronger branch to Block 4B (completing the overtake!)
    simulator.add_block("BLOCK_4B", "BLOCK_3B", difficulty=100, transactions=["Tx: Eve to Grace 0.4 BTC"])

As shown, the script properly tracks blocks, computes branches, locates the correct common ancestor block, executes clean state rollbacks, and shifts node states seamlessly during reorgs.