Bitcoin Protocol Explained — Transactions, Blocks, and the Network

# Bitcoin transaction structure (simplified)
from dataclasses import dataclass, field
from typing import List
import hashlib

@dataclass
class TransactionInput:
    previous_tx_hash: str       # which UTXO we're spending
    previous_output_index: int  # which output in that tx
    script_sig: str             # unlocking script (signature)
    sequence: int = 0xFFFFFFFF

@dataclass
class TransactionOutput:
    value: int                  # amount in satoshis (1 BTC = 100M sat)
    script_pubkey: str          # locking script (recipient conditions)

@dataclass
class Transaction:
    version: int = 1
    inputs: List[TransactionInput] = field(default_factory=list)
    outputs: List[TransactionOutput] = field(default_factory=list)
    locktime: int = 0

    def txid(self) -> str:
        """Compute the transaction ID (double SHA-256 of serialized tx)."""
        serialized = f"{self.version}{len(self.inputs)}{self.locktime}"
        first_hash = hashlib.sha256(serialized.encode()).digest()
        return hashlib.sha256(first_hash).hexdigest()

# Example: Alice pays Bob 12 BTC
tx = Transaction(
    inputs=[TransactionInput(
        previous_tx_hash="a1b2c3d4e5f6...",
        previous_output_index=0,
        script_sig="<Alice's signature> <Alice's public key>]
    )],
    outputs=[
        TransactionOutput(value=1_200_000_000, "# 12 BTC in satoshis
                         script_pubkey="OP_DUP OP_HASH160 <Bob's hash> OP_EQUALVERIFY OP_CHECKSIG")",
        TransactionOutput(value=299_000_000,   # 2.99 BTC change
                         script_pubkey="OP_DUP OP_HASH160 <Alice's hash> OP_EQUALVERIFY OP_CHECKSIG")
    ]
)
print(f"Transaction ID: {tx.txid()}")
print(f"Total input: {sum(i.previous_output_index for i in tx.inputs)}")  # simplified

# Bitcoin Script simulation in Python
def evaluate_script(locking_script: List[str], unlocking_script: List[str]) -> bool:
    """Simulate Bitcoin Script execution for a P2PKH transaction."""
    import hashlib

    stack = []

    # Concatenate unlocking + locking scripts
    full_script = unlocking_script + locking_script

    i = 0
    while i < len(full_script):
        op = full_script[i]

        if op == "OP_DUP":
            # Duplicate top of stack
            if len(stack) >= 1:
                stack.append(stack[-1])

        elif op == "OP_HASH160":
            # RIPEMD-160(SHA-256(x))
            if len(stack) >= 1:
                data = stack.pop().encode()
                h = hashlib.new('ripemd160', hashlib.sha256(data).digest()).digest()
                stack.append(h.hex())

        elif op == "OP_EQUALVERIFY":
            # Check top two are equal, then pop both
            if len(stack) >= 2:
                a = stack.pop()
                b = stack.pop()
                if a != b:
                    return False

        elif op == "OP_CHECKSIG":
            # Simplified signature check (always returns True for demo)
            if len(stack) >= 2:
                sig = stack.pop()
                pubkey = stack.pop()
                stack.append(1)  # assume valid signature

        else:
            # It's a data push
            stack.append(op)

        i += 1

    # Script is valid if top of stack is non-zero
    return len(stack) > 0 and stack[-1] == 1

# Example: Valid P2PKH spend
locking = [
    "OP_DUP", "OP_HASH160", "abc123def456...", "OP_EQUALVERIFY", "OP_CHECKSIG]
]
unlocking = [
    "3045022100...",  # signature
    "04a1b2c3..."     # public key
]

result = evaluate_script(locking, unlocking)
print(f"Script valid: {result}")

# Example: Invalid signature (script fails)
unlocking_bad = [
    "3045022100...BAD",
    "04a1b2c3...]
]
result_bad = evaluate_script(locking, unlocking_bad)
print(f"Bad script valid: {result_bad}")

# Simplified Bitcoin block propagation simulation
import random
import time
from typing import Set, List

class BitcoinNode:
    def __init__(self, node_id: str):
        self.node_id = node_id
        self.peers: Set['BitcoinNode'] = set()
        self.known_blocks: Set[str] = set()
        self.known_transactions: Set[str] = set()
        self.mempool: List[dict] = []

    def connect(self, peer: 'BitcoinNode'):
        self.peers.add(peer)
        peer.peers.add(self)

    def broadcast_transaction(self, tx: dict):
        """Gossip a transaction to all connected peers."""
        tx_id = tx.get('txid', '')
        if tx_id in self.known_transactions:
            return  # already seen
        self.known_transactions.add(tx_id)
        self.mempool.append(tx)

        for peer in self.peers:
            peer.receive_transaction(tx, self)

    def receive_transaction(self, tx: dict, sender: 'BitcoinNode'):
        """Receive a transaction from a peer and propagate."""
        tx_id = tx.get('txid', '')
        if tx_id in self.known_transactions:
            return
        self.known_transactions.add(tx_id)
        self.mempool.append(tx)

        # Relay to all peers except the sender
        for peer in self.peers:
            if peer != sender:
                peer.receive_transaction(tx, self)

    def mine_block(self, difficulty: int = 4) -> dict:
        """Create a new block from mempool transactions (simplified PoW)."""
        import hashlib

        block = {
            'height': len(self.known_blocks),
            'transactions': self.mempool[:3],  # include up to 3 txs
            'nonce': 0,
            'previous_block': list(self.known_blocks)[-1] if self.known_blocks else '0' * 64
        }

        # Simplified PoW
        prefix = "0" * difficulty
        while True:
            data = f"{block['height']}{block['transactions']}{block['nonce']}"
            hash_result = hashlib.sha256(data.encode()).hexdigest()
            if hash_result.startswith(prefix):
                block['hash'] = hash_result
                break
            block['nonce'] += 1

        self.known_blocks.add(block['hash'])
        self.mempool = []
        return block

# Simulate a small network
alice = BitcoinNode("Alice")
bob = BitcoinNode("Bob")
carol = BitcoinNode("Carol")
dave = BitcoinNode("Dave")

alice.connect(bob)
bob.connect(carol)
carol.connect(dave)

# Alice creates and broadcasts a transaction
tx1 = {"txid": "abc123", "from": "Alice", "to": "Bob", "amount": 5}
alice.broadcast_transaction(tx1)

print(f"Bob knows {len(bob.known_transactions)} transaction(s): {bob.known_transactions}")
print(f"Dave knows {len(dave.known_transactions)} transaction(s): {dave.known_transactions}")

# Carol mines a block
block = carol.mine_block()
print(f"Carol mined block #{block['height']} with hash {block['hash'][:16]}...")

# Bitcoin difficulty adjustment simulation
import math

def calculate_new_difficulty(
    time_taken_seconds: float,
    target_block_time: float = 600,  # 10 minutes in seconds
    adjustment_window: int = 2016
) -> float:
    """
    Calculate the new difficulty target based on the time taken to mine
    the last 2,016 blocks.
    """
    expected_time = adjustment_window * target_block_time

    # Clamp the ratio to 4x or 1/4x max adjustment
    ratio = time_taken_seconds / expected_time
    ratio = max(0.25, min(ratio, 4.0))

    new_difficulty = ratio
    return new_difficulty

# Scenario 1: Blocks coming too fast (more miners joined)
fast_time = 2016 * 480  # 8 min avg instead of 10
new_diff_fast = calculate_new_difficulty(fast_time)
print(f"Fast scenario (8 min avg): difficulty multiplier = {new_diff_fast:.4f}")
print(f"  Difficulty INCREASES by {((1 - new_diff_fast) * 100):.1f}%")

# Scenario 2: Blocks coming too slow (miners left)
slow_time = 2016 * 900  # 15 min avg instead of 10
new_diff_slow = calculate_new_difficulty(slow_time)
print(f"Slow scenario (15 min avg): difficulty multiplier = {new_diff_slow:.4f}")
print(f"  Difficulty DECREASES by {((new_diff_slow - 1) * 100):.1f}%")

# Scenario 3: Extreme case (capped at 4x)
extreme_time = 2016 * 3600  # 60 min avg
new_diff_extreme = calculate_new_difficulty(extreme_time)
print(f"Extreme scenario (60 min avg): difficulty multiplier = {new_diff_extreme:.4f} (capped)")