Key Code Examples
1. Basic Blockchain Implementation
import hashlib
import json
import time
from typing import List, Optional
from dataclasses import dataclass, asdict
@dataclass
class Transaction:
sender: str
recipient: str
amount: float
timestamp: float = None
signature: str = ""
def __post_init__(self):
if self.timestamp is None:
self.timestamp = time.time()
def to_dict(self):
return asdict(self)
def to_json(self):
return json.dumps(self.to_dict(), sort_keys=True)
@dataclass
class Block:
index: int
transactions: List[Transaction]
timestamp: float
previous_hash: str
nonce: int = 0
hash: str = ""
def compute_hash(self) -> str:
"""Calculate SHA-256 hash of the block."""
block_data = json.dumps({
"index": self.index,
"transactions": [t.to_dict() for t in self.transactions],
"timestamp": self.timestamp,
"previous_hash": self.previous_hash,
"nonce": self.nonce
}, sort_keys=True)
return hashlib.sha256(block_data.encode()).hexdigest()
def mine_block(self, difficulty: int = 4):
"""Proof of Work: Find hash with leading zeros."""
target = "0" * difficulty
print(f"Mining block {self.index}...")
start = time.time()
while True:
self.hash = self.compute_hash()
if self.hash.startswith(target):
elapsed = time.time() - start
print(f"Block {self.index} mined! Hash: {self.hash[:16]}...")
print(f"Nonce: {self.nonce}, Time: {elapsed:.2f}s")
break
self.nonce += 1
class Blockchain:
def __init__(self, difficulty=4):
self.chain: List[Block] = []
self.pending_transactions: List[Transaction] = []
self.difficulty = difficulty
self.mining_reward = 50.0
self.create_genesis_block()
def create_genesis_block(self):
"""Create the first block in the chain."""
genesis = Block(
index=0,
transactions=[],
timestamp=time.time(),
previous_hash="0" * 64
)
genesis.mine_block(self.difficulty)
self.chain.append(genesis)
def get_latest_block(self) -> Block:
return self.chain[-1]
def add_transaction(self, transaction: Transaction) -> bool:
"""Add a transaction to pending pool."""
if not transaction.sender or not transaction.recipient:
return False
if transaction.amount <= 0:
return False
self.pending_transactions.append(transaction)
return True
def mine_pending_transactions(self, miner_address: str):
"""Mine pending transactions into a new block."""
# Reward miner
reward_tx = Transaction(
sender="system",
recipient=miner_address,
amount=self.mining_reward
)
transactions = self.pending_transactions + [reward_tx]
new_block = Block(
index=len(self.chain),
transactions=transactions,
timestamp=time.time(),
previous_hash=self.get_latest_block().hash
)
new_block.mine_block(self.difficulty)
self.chain.append(new_block)
self.pending_transactions = []
return new_block
def is_chain_valid(self) -> bool:
"""Verify blockchain integrity."""
for i in range(1, len(self.chain)):
current = self.chain[i]
previous = self.chain[i-1]
# Verify hash
if current.hash != current.compute_hash():
print(f"Block {i}: Invalid hash!")
return False
# Verify chain linkage
if current.previous_hash != previous.hash:
print(f"Block {i}: Chain broken!")
return False
# Verify proof of work
if not current.hash.startswith("0" * self.difficulty):
print(f"Block {i}: Invalid proof of work!")
return False
return True
def get_balance(self, address: str) -> float:
"""Calculate balance for an address."""
balance = 0.0
for block in self.chain:
for tx in block.transactions:
if tx.sender == address:
balance -= tx.amount
if tx.recipient == address:
balance += tx.amount
return balance
# Demo
blockchain = Blockchain(difficulty=3)
print("Genesis block created!")
blockchain.add_transaction(Transaction("Alice", "Bob", 50.0))
blockchain.add_transaction(Transaction("Bob", "Charlie", 25.0))
blockchain.mine_pending_transactions("Miner1")
print(f"\nBlockchain valid: {blockchain.is_chain_valid()}")
print(f"Alice's balance: {blockchain.get_balance('Alice')}")
print(f"Bob's balance: {blockchain.get_balance('Bob')}")
print(f"Miner1's balance: {blockchain.get_balance('Miner1')}")
2. Merkle Tree Implementation
import hashlib
from typing import List, Optional
class MerkleTree:
"""Merkle Tree for efficient transaction verification."""
def __init__(self, data: List[str]):
self.leaves = [self._hash(d) for d in data]
self.levels = [self.leaves]
self.root = self._build_tree(self.leaves)
@staticmethod
def _hash(data: str) -> str:
return hashlib.sha256(data.encode()).hexdigest()
@staticmethod
def _hash_pair(a: str, b: str) -> str:
combined = a + b
return hashlib.sha256(combined.encode()).hexdigest()
def _build_tree(self, leaves: List[str]) -> str:
"""Build tree bottom-up and return root hash."""
if not leaves:
return self._hash("")
current_level = leaves
self.levels.append(current_level)
while len(current_level) > 1:
next_level = []
# Pair adjacent hashes
for i in range(0, len(current_level), 2):
left = current_level[i]
right = current_level[i + 1] if i + 1 < len(current_level) else left
next_level.append(self._hash_pair(left, right))
current_level = next_level
self.levels.append(current_level)
return current_level[0]
def get_proof(self, index: int) -> List[tuple]:
"""Get Merkle proof for a leaf at given index."""
proof = []
for level in self.levels[:-1]: # Exclude root
if index % 2 == 0:
sibling_idx = index + 1
if sibling_idx < len(level):
proof.append(("right", level[sibling_idx]))
else:
sibling_idx = index - 1
proof.append(("left", level[sibling_idx]))
index //= 2
return proof
@staticmethod
def verify_proof(leaf: str, proof: List[tuple], root: str) -> bool:
"""Verify a Merkle proof."""
current = hashlib.sha256(leaf.encode()).hexdigest()
for direction, sibling_hash in proof:
if direction == "left":
current = hashlib.sha256((sibling_hash + current).encode()).hexdigest()
else:
current = hashlib.sha256((current + sibling_hash).encode()).hexdigest()
return current == root
# Example: Verify transactions in a block
transactions = [
"Alice -> Bob: 50 BTC",
"Bob -> Charlie: 25 BTC",
"Charlie -> Dave: 10 BTC",
"Dave -> Eve: 5 BTC"
]
tree = MerkleTree(transactions)
print(f"Merkle Root: {tree.root}")
# Prove that transaction 1 is in the tree
proof = tree.get_proof(1)
print(f"Proof for tx[1]: {proof}")
is_valid = MerkleTree.verify_proof(transactions[1], proof, tree.root)
print(f"Verification: {is_valid}") # True
3. Digital Signatures (RSA)
import hashlib
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import hashes, serialization
class DigitalWallet:
"""Blockchain wallet with RSA digital signatures."""
def __init__(self):
# Generate RSA key pair
self.private_key = rsa.generate_private_key(
public_exponent=65537,
key_size=2048
)
self.public_key = self.private_key.public_key()
self.address = self._generate_address()
def _generate_address(self) -> str:
"""Generate address from public key hash."""
pub_key_bytes = self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
return hashlib.sha256(pub_key_bytes).hexdigest()[:40]
def sign_transaction(self, transaction_data: str) -> bytes:
"""Sign transaction data with private key."""
signature = self.private_key.sign(
transaction_data.encode(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return signature
@staticmethod
def verify_signature(public_key, transaction_data: str,
signature: bytes) -> bool:
"""Verify a transaction signature."""
try:
public_key.verify(
signature,
transaction_data.encode(),
padding.PSS(
mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH
),
hashes.SHA256()
)
return True
except Exception:
return False
def create_transaction(self, recipient: str, amount: float,
nonce: int = 0) -> dict:
"""Create and sign a transaction."""
tx_data = {
"from": self.address,
"to": recipient,
"amount": amount,
"nonce": nonce,
"timestamp": time.time()
}
tx_json = json.dumps(tx_data, sort_keys=True)
signature = self.sign_transaction(tx_json)
return {
**tx_data,
"signature": signature.hex(),
"public_key": self.public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
).hex()
}
# Usage
wallet = DigitalWallet()
print(f"Wallet address: {wallet.address}")
tx = wallet.create_transaction(recipient="1abc23...", amount=25.0)
print(f"Transaction: {json.dumps(tx, indent=2)}")
# Verify
is_valid = DigitalWallet.verify_signature(
wallet.private_key.public_key(),
json.dumps({k: v for k, v in tx.items() if k != 'signature'}, sort_keys=True),
bytes.fromhex(tx['signature'])
)
print(f"Signature valid: {is_valid}")
4. Smart Contract (Solidity)
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.20;
/// @title Simple Storage Contract
contract SimpleStorage {
uint256 private value;
address public owner;
event ValueChanged(address indexed who, uint256 newValue);
constructor() {
owner = msg.sender;
}
modifier onlyOwner() {
require(msg.sender == owner, "Not the owner");
_;
}
function setValue(uint256 _value) public {
value = _value;
emit ValueChanged(msg.sender, _value);
}
function getValue() public view returns (uint256) {
return value;
}
}
/// @title Token Contract (ERC-20-like)
contract SimpleToken {
string public name = "EduVerseToken";
string public symbol = "EVT";
uint8 public decimals = 18;
uint256 public totalSupply;
mapping(address => uint256) public balanceOf;
mapping(address => mapping(address => uint256)) public allowance;
event Transfer(address indexed from, address indexed to, uint256 value);
event Approval(address indexed owner, address indexed spender, uint256 value);
constructor(uint256 initialSupply) {
totalSupply = initialSupply * 10 ** uint256(decimals);
balanceOf[msg.sender] = totalSupply;
emit Transfer(address(0), msg.sender, totalSupply);
}
function transfer(address to, uint256 amount) public returns (bool) {
require(balanceOf[msg.sender] >= amount, "Insufficient balance");
require(to != address(0), "Invalid recipient");
balanceOf[msg.sender] -= amount;
balanceOf[to] += amount;
emit Transfer(msg.sender, to, amount);
return true;
}
function approve(address spender, uint256 amount) public returns (bool) {
allowance[msg.sender][spender] = amount;
emit Approval(msg.sender, spender, amount);
return true;
}
function transferFrom(address from, address to, uint256 amount) public returns (bool) {
require(balanceOf[from] >= amount, "Insufficient balance");
require(allowance[from][msg.sender] >= amount, "Allowance exceeded");
balanceOf[from] -= amount;
balanceOf[to] += amount;
allowance[from][msg.sender] -= amount;
emit Transfer(from, to, amount);
return true;
}
}
/// @title Voting Contract
contract Voting {
struct Proposal {
string name;
uint256 voteCount;
}
mapping(address => bool) public hasVoted;
Proposal[] public proposals;
address public chairperson;
constructor(string[] memory proposalNames) {
chairperson = msg.sender;
for (uint i = 0; i < proposalNames.length; i++) {
proposals.push(Proposal({name: proposalNames[i], voteCount: 0}));
}
}
function vote(uint256 proposal) public {
require(!hasVoted[msg.sender], "Already voted");
require(proposal < proposals.length, "Invalid proposal");
hasVoted[msg.sender] = true;
proposals[proposal].voteCount++;
}
function winningProposal() public view returns (uint256) {
uint256 highest = 0;
uint256 winner = 0;
for (uint i = 0; i < proposals.length; i++) {
if (proposals[i].voteCount > highest) {
highest = proposals[i].voteCount;
winner = i;
}
}
return winner;
}
}
5. Proof of Stake Consensus
import hashlib
import random
import time
from typing import List, Dict
class Validator:
def __init__(self, address: str, stake: float):
self.address = address
self.stake = stake
self.reputation = 100
@property
def weight(self) -> float:
"""Selection weight based on stake and reputation."""
return self.stake * (self.reputation / 100)
class ProofOfStake:
"""Proof of Stake consensus mechanism."""
def __init__(self):
self.validators: Dict[str, Validator] = {}
self.stake_pool = 0.0
self.min_stake = 1000.0
self.slash_conditions = ["double_sign", "downtime"]
def register_validator(self, address: str, stake: float) -> bool:
"""Register a new validator with stake."""
if stake < self.min_stake:
print(f"Stake {stake} below minimum {self.min_stake}")
return False
self.validators[address] = Validator(address, stake)
self.stake_pool += stake
print(f"Validator {address[:8]}... registered with stake {stake}")
return True
def select_validator(self, seed: str) -> str:
"""Pseudo-random validator selection weighted by stake."""
if not self.validators:
raise ValueError("No validators registered")
# Use seed for deterministic selection
random.seed(seed)
total_weight = sum(v.weight for v in self.validators.values())
# Weighted random selection
selection = random.uniform(0, total_weight)
cumulative = 0
for address, validator in self.validators.items():
cumulative += validator.weight
if cumulative >= selection:
return address
return list(self.validators.keys())[-1]
def validate_block(self, validator_address: str,
block_hash: str) -> dict:
"""Validator proposes/validates a block."""
if validator_address not in self.validators:
return {"valid": False, "reason": "Not a validator"}
validator = self.validators[validator_address]
# Check if validator is selected (probabilistic)
selected = self.select_validator(block_hash)
if selected != validator_address:
return {"valid": False, "reason": "Validator not selected"}
# Simulate validation
time.sleep(0.01) # Block validation time
# Reward validator
reward = 10.0 # Block reward
validator.stake += reward
return {
"valid": True,
"validator": validator_address,
"reward": reward,
"timestamp": time.time()
}
def slash_validator(self, address: str, reason: str):
"""Penalize misbehaving validator."""
if address in self.validators:
validator = self.validators[address]
slash_amount = validator.stake * 0.05 # 5% slashing
validator.stake -= slash_amount
validator.reputation -= 20
self.stake_pool -= slash_amount
print(f"Validator {address[:8]} slashed {slash_amount} for {reason}")
def get_staking_apy(self, address: str) -> float:
"""Calculate estimated annual yield."""
if address not in self.validators:
return 0.0
validator = self.validators[address]
return (validator.stake / self.stake_pool) * 100
# Demo
pos = ProofOfStake()
pos.register_validator("Validator_A", 50000)
pos.register_validator("Validator_B", 30000)
pos.register_validator("Validator_C", 20000)
# Select validator for block
block_hash = hashlib.sha256(b"block_123").hexdigest()
selected = pos.select_validator(block_hash)
print(f"Selected validator: {selected[:8]}...")
result = pos.validate_block(selected, block_hash)
print(f"Validation result: {result}")
6. DApp Interaction (Web3.py)
from web3 import Web3
import json
class BlockchainDApp:
"""Interact with Ethereum blockchain using Web3."""
def __init__(self, provider_url: str):
self.w3 = Web3(Web3.HTTPProvider(provider_url))
if not self.w3.is_connected():
raise ConnectionError("Cannot connect to Ethereum node")
print(f"Connected to Ethereum (Chain ID: {self.w3.eth.chain_id})")
def get_balance(self, address: str) -> float:
"""Get ETH balance for an address."""
balance_wei = self.w3.eth.get_balance(address)
return self.w3.from_wei(balance_wei, 'ether')
def deploy_contract(self, contract_interface: dict,
private_key: str) -> str:
"""Deploy a smart contract."""
account = self.w3.eth.account.from_key(private_key)
Contract = self.w3.eth.contract(
abi=contract_interface['abi'],
bytecode=contract_interface['bin']
)
# Build and sign transaction
tx = Contract.constructor().build_transaction({
'from': account.address,
'nonce': self.w3.eth.get_transaction_count(account.address),
'gas': 2000000,
'gasPrice': self.w3.eth.gas_price,
})
signed = self.w3.eth.account.sign_transaction(tx, private_key)
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
print(f"Contract deployed at: {receipt.contractAddress}")
return receipt.contractAddress
def call_contract_function(self, contract_address: str,
abi: list, function_name: str,
args: list = None,
private_key: str = None) -> any:
"""Read or write contract state."""
contract = self.w3.eth.contract(
address=contract_address,
abi=abi
)
func = getattr(contract.functions, function_name)
if args:
func = func(*args)
if private_key:
# Write transaction
account = self.w3.eth.account.from_key(private_key)
tx = func.build_transaction({
'from': account.address,
'nonce': self.w3.eth.get_transaction_count(account.address),
'gas': 200000,
'gasPrice': self.w3.eth.gas_price,
})
signed = self.w3.eth.account.sign_transaction(tx, private_key)
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
return receipt
else:
# Read-only call
return func.call()
def transfer_tokens(self, token_address: str, abi: list,
private_key: str, to_address: str,
amount: int) -> dict:
"""Transfer ERC-20 tokens."""
account = self.w3.eth.account.from_key(private_key)
contract = self.w3.eth.contract(address=token_address, abi=abi)
tx = contract.functions.transfer(to_address, amount).build_transaction({
'from': account.address,
'nonce': self.w3.eth.get_transaction_count(account.address),
'gas': 100000,
'gasPrice': self.w3.eth.gas_price,
})
signed = self.w3.eth.account.sign_transaction(tx, private_key)
tx_hash = self.w3.eth.send_raw_transaction(signed.rawTransaction)
receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
return {"tx_hash": tx_hash.hex(), "status": receipt.status}
# Usage
# dapp = BlockchainDApp("https://mainnet.infura.io/v3/YOUR_KEY")
# balance = dapp.get_balance("0x...")
# print(f"Balance: {balance} ETH")
7. Hashcash Proof of Work
import hashlib
import time
class HashCash:
"""Hashcash-style proof of work (used in Bitcoin)."""
def __init__(self, bits=20):
self.bits = bits # Number of leading zero bits required
def _count_leading_zeros(self, hash_hex: str) -> int:
"""Count leading zero bits in hash."""
count = 0
for char in hash_hex:
value = int(char, 16)
if value == 0:
count += 4
else:
# Count leading zeros in nibble
for bit in range(3, -1, -1):
if value & (1 << bit):
return count
count += 1
return count
return count
def mint(self, resource: str, timestamp=None) -> dict:
"""Find a valid nonce (mint a hashcash stamp)."""
if timestamp is None:
timestamp = int(time.time())
challenge = f"{resource}:{timestamp}"
nonce = 0
print(f"Minting hashcash ({self.bits}-bit difficulty)...")
start = time.time()
while True:
attempt = f"{challenge}:{nonce}"
hash_result = hashlib.sha256(attempt.encode()).hexdigest()
if self._count_leading_zeros(hash_result) >= self.bits:
elapsed = time.time() - start
print(f"Found nonce: {nonce} in {elapsed:.2f}s")
print(f"Hash: {hash_result[:16]}...")
return {
"resource": resource,
"timestamp": timestamp,
"nonce": nonce,
"hash": hash_result,
"bits": self.bits
}
nonce += 1
@staticmethod
def verify(stamp: dict) -> bool:
"""Verify a hashcash stamp."""
challenge = f"{stamp['resource']}:{stamp['timestamp']}"
attempt = f"{challenge}:{stamp['nonce']}"
hash_result = hashlib.sha256(attempt.encode()).hexdigest()
# Check difficulty
zeros = 0
for char in hash_result:
value = int(char, 16)
if value == 0:
zeros += 4
else:
for bit in range(3, -1, -1):
if value & (1 << bit):
break
zeros += 1
break
return zeros >= stamp['bits'] and hash_result == stamp['hash']
# Example: Email anti-spam ( Hashcash use case )
hashcash = HashCash(bits=16)
stamp = hashcash.mint(resource="user@example.com")
print(f"\nStamp: {stamp}")
is_valid = HashCash.verify(stamp)
print(f"Verification: {is_valid}") # True
8. Decentralized Identity (DID)
import hashlib
import json
import time
from typing import Dict, List
class DecentralizedIdentity:
"""Self-sovereign identity on blockchain."""
def __init__(self, did_method="did:edu"):
self.did_method = did_method
self.identities = {}
def create_identity(self, public_key: str, metadata: dict = None) -> dict:
"""Create a new DID (Decentralized Identifier)."""
# Generate DID
did_hash = hashlib.sha256(public_key.encode()).hexdigest()[:32]
did = f"{self.did_method}:{did_hash}"
identity = {
"did": did,
"publicKey": public_key,
"metadata": metadata or {},
"created": time.time(),
"controls": [],
"services": []
}
self.identities[did] = identity
return identity
def add_controller(self, did: str, controller_did: str):
"""Add another controller for the identity."""
if did in self.identities:
self.identities[did]["controls"].append({
"did": controller_did,
"granted": time.time()
})
def add_service(self, did: str, service_type: str,
endpoint: str):
"""Add a service endpoint to the identity."""
if did in self.identities:
self.identities[did]["services"].append({
"type": service_type,
"endpoint": endpoint,
"added": time.time()
})
def resolve(self, did: str) -> dict:
"""Resolve a DID to its document."""
return self.identities.get(did)
def create_credential(self, issuer_did: str, subject_did: str,
claims: dict) -> dict:
"""Issue a verifiable credential."""
credential = {
"@context": "https://www.w3.org/2018/credentials/v1",
"id": hashlib.sha256(
f"{issuer_did}:{subject_did}:{time.time()}".encode()
).hexdigest(),
"issuer": issuer_did,
"issuanceDate": time.time(),
"credentialSubject": {
"id": subject_did,
**claims
},
"proof": {
"type": "Ed25519Signature2020",
"issuer": issuer_did,
"issued": time.time()
}
}
# In production: sign the credential with issuer's private key
return credential
# Usage
did_system = DecentralizedIdentity()
# Create identity
identity = did_system.create_identity(
public_key="0xABC123...",
metadata={"name": "Alice Smith", "email": "alice@example.com"}
)
print(f"DID: {identity['did']}")
# Add service
did_system.add_service(identity['did'], "Messaging",
"https://messaging.eduverse.com/alice")
# Issue credential
credential = did_system.create_credential(
issuer_did="did:edu:issuer123",
subject_did=identity['did'],
claims={"degree": "BSc Computer Science", "university": "EduVerse University"}
)
print(f"Credential: {json.dumps(credential, indent=2)}")
9. Tokenomics — Staking Rewards Calculator
class StakingCalculator:
"""Calculate staking rewards and tokenomics metrics."""
def __init__(self, total_supply: float, annual_inflation: float = 0.05):
self.total_supply = total_supply
self.annual_inflation = annual_inflation
def calculate_staking_apy(self, total_staked: float,
validator_fee: float = 0.10) -> dict:
"""Calculate Annual Percentage Yield for staking."""
if total_staked <= 0:
return {"error": "No tokens staked"}
# Issuance per year
annual_issuance = self.total_supply * self.annual_inflation
# Proportional to stake
staking_ratio = total_staked / self.total_supply
# Gross rewards
gross_apy = annual_issuance / total_staked if total_staked > 0 else 0
# Net after validator fee
net_apy = gross_apy * (1 - validator_fee)
return {
"gross_apy": f"{gross_apy*100:.2f}%",
"net_apy": f"{net_apy*100:.2f}%",
"staking_ratio": f"{staking_ratio*100:.2f}%",
"annual_issuance": f"{annual_issuance:,.0f}",
}
def compound_rewards(self, principal: float, apy: float,
years: int, compound_frequency: int = 365) -> dict:
"""Calculate compound staking rewards."""
rate = apy / compound_frequency
periods = years * compound_frequency
final_amount = principal * (1 + rate) ** periods
total_rewards = final_amount - principal
yearly_breakdown = []
for year in range(1, years + 1):
year_amount = principal * (1 + rate) ** (year * compound_frequency)
yearly_breakdown.append({
"year": year,
"balance": f"{year_amount:,.2f}",
"cumulative_rewards": f"{year_amount - principal:,.2f}"
})
return {
"principal": f"{principal:,.2f}",
"apy": f"{apy*100:.2f}%",
"years": years,
"final_amount": f"{final_amount:,.2f}",
"total_rewards": f"{total_rewards:,.2f}",
"roi": f"{(total_rewards/principal)*100:.2f}%",
"yearly_breakdown": yearly_breakdown
}
def slashing_risk(self, validator_uptime: float,
stake: float, slash_rate: float = 0.05) -> dict:
"""Calculate risk-adjusted returns."""
uptime_factor = validator_uptime / 100
expected_slash = stake * slash_rate * (1 - uptime_factor)
effective_yield = (stake * 0.07 * uptime_factor) - expected_slash
return {
"uptime": f"{validator_uptime}%",
"expected_slash_risk": f"{expected_slash:,.2f}",
"effective_annual_return": f"{effective_yield:,.2f}",
"roi": f"{(effective_yield/stake)*100:.2f}%"
}
# Example
calc = StakingCalculator(total_supply=100_000_000, annual_inflation=0.05)
# Calculate APY with 60% staking ratio
result = calc.calculate_staking_apy(total_staked=60_000_000)
print("Staking APY:", result)
# Compound rewards for 10,000 tokens over 5 years at 7% APY
compound = calc.compound_rewards(principal=10000, apy=0.07, years=5)
print(f"\n5-Year Projection:")
print(f" Final amount: {compound['final_amount']}")
print(f" Total rewards: {compound['total_rewards']}")
print(f" ROI: {compound['roi']}")
class CrossChainBridge:
"""Cross-chain asset bridge using lock-and-mint mechanism."""
def __init__(self):
self.supported_chains = {}
self.bridge_contracts = {}
self.wrapped_assets = {}
def register_chain(self, chain_id: str, bridge_address: str,
confirmation_blocks: int):
"""Register a supported blockchain."""
self.supported_chains[chain_id] = {
"bridge_contract": bridge_address,
"confirmations": confirmation_blocks,
"status": "active"
}
print(f"Registered chain: {chain_id}")
def lock_assets(self, chain_id: str, asset: str, amount: float,
sender: str, target_chain: str,
recipient: str) -> dict:
"""Lock assets on source chain."""
if chain_id not in self.supported_chains:
raise ValueError(f"Chain {chain_id} not supported")
tx_hash = hashlib.sha256(
f"{chain_id}:{sender}:{amount}:{time.time()}".encode()
).hexdigest()
lock_record = {
"source_chain": chain_id,
"target_chain": target_chain,
"asset": asset,
"amount": amount,
"sender": sender,
"recipient": recipient,
"tx_hash": tx_hash,
"status": "locked",
"timestamp": time.time(),
"confirmations": 0
}
print(f"Locked {amount} {asset} on {chain_id}")
print(f"Waiting for {self.supported_chains[chain_id]['confirmations']} confirmations...")
return lock_record
def mint_wrapped(self, lock_record: dict) -> str:
"""Mint wrapped tokens on target chain."""
wrapped_asset = f"W{lock_record['asset']}"
target_chain = lock_record['target_chain']
mint_tx = hashlib.sha256(
f"mint:{target_chain}:{lock_record['tx_hash']}".encode()
).hexdigest()
self.wrapped_assets[mint_tx] = {
"asset": wrapped_asset,
"amount": lock_record['amount'],
"recipient": lock_record['recipient'],
"source_tx": lock_record['tx_hash']
}
print(f"Minted {lock_record['amount']} {wrapped_asset} on {target_chain}")
return mint_tx
def burn_and_release(self, wrapped_tx: str) -> dict:
"""Burn wrapped tokens and release on source chain."""
if wrapped_tx not in self.wrapped_assets:
raise ValueError("Invalid wrapped asset")
record = self.wrapped_assets[wrapped_tx]
# Burn transaction
burn_tx = hashlib.sha256(
f"burn:{wrapped_tx}:{time.time()}".encode()
).hexdigest()
# Release on source chain
release_tx = hashlib.sha256(
f"release:{burn_tx}".encode()
).hexdigest()
result = {
"burn_tx": burn_tx,
"release_tx": release_tx,
"asset": record['asset'].lstrip('W'),
"amount": record['amount'],
"status": "released"
}
print(f"Burned {record['asset']}, released {result['asset']} on source chain")
return result
# Usage
bridge = CrossChainBridge()
bridge.register_chain("ethereum", "0xBridgeEth...", 12)
bridge.register_chain("polygon", "0xBridgePoly...", 20)
# Bridge 100 USDC from Ethereum to Polygon
lock = bridge.lock_assets("ethereum", "USDC", 100,
"0xSender...", "polygon", "0xRecipient...")
wrapped = bridge.mint_wrapped(lock)
# bridge.burn_and_release(wrapped)
Core Concepts
What is Blockchain?
Definition: A distributed, immutable ledger maintained by a network of computers (nodes). Each block contains transactions, a timestamp, and the hash of the previous block.
Key Properties:
- Decentralized: No single authority controls the network
- Immutable: Once recorded, data cannot be altered
- Transparent: All transactions are publicly verifiable
- Secure: Cryptographic hashing links blocks together
Consensus Mechanisms
- Proof of Work (PoW): Miners solve computational puzzles. Secure but energy-intensive. Used by Bitcoin.
- Proof of Stake (PoS): Validators stake tokens as collateral. Energy-efficient. Used by Ethereum 2.0.
- Delegated PoS (DPoS): Token holders vote for validators. Fast but more centralized.
- Proof of Authority (PoA): Pre-approved validators. Used in private/consortium chains.
Cryptographic Hashing
SHA-256: Produces 256-bit (64 hex char) output. Used in Bitcoin mining and block hashing.
Properties: Deterministic, fast, preimage-resistant, collision-resistant, avalanche effect.
Keccak-256: Ethereum's hash function. Used for addresses and smart contract verification.
Smart Contracts
Definition: Self-executing programs stored on the blockchain that automatically enforce rules when conditions are met.
- Turing Complete: Can compute anything given enough resources (gas limit)
- Deterministic: Same input always produces same output
- Immutable: Deployed code cannot be changed
- Examples: DeFi protocols, NFTs, DAOs, supply chain
Blockchain Types
- Public: Open to anyone (Bitcoin, Ethereum). Fully decentralized.
- Private: Restricted access, single organization. Enterprise use.
- Consortium: Multiple organizations share control (Hyperledger Fabric).
- Layer 2: Solutions built on top of Layer 1 for scaling (Lightning, Rollups).
DeFi (Decentralized Finance)
Definition: Financial services built on blockchain without intermediaries.
- DEX (Decentralized Exchange): Automated market makers like Uniswap
- Lending/Borrowing: Aave, Compound — earn interest on deposits
- Stablecoins: USDC, DAI — price-stable tokens pegged to fiat
- Yield Farming: Provide liquidity for token rewards
- Flash Loans: Uncollateralized loans in single transaction
NFTs (Non-Fungible Tokens)
Definition: Unique digital assets verified on blockchain, each with distinct properties.
- ERC-721: Standard for unique tokens (CryptoPunks, Bored Apes)
- ERC-1155: Semi-fungible tokens (gaming items)
- Use Cases: Digital art, music, gaming, real estate, identity
- Metadata: Image URI, properties, provenance stored on-chain/IPFS
Scalability Trilemma
Blockchain Trilemma: Achieving decentralization, security, and scalability simultaneously is extremely difficult. Most blockchains optimize for 2 of 3.
- Layer 1 Solutions: Sharding, consensus improvements, larger blocks
- Layer 2 Solutions: Rollups (Optimistic, ZK), state channels, sidechains
- Current Status: Ethereum L2s process 10-100x more TPS than mainnet
Worked Examples
Example 1: Mine a Block Step-by-Step
Problem: Mine a block with difficulty=3 (hash must start with "000").
Solution:
import hashlib
import time
def mine_block(block_data, difficulty=3):
target = "0" * difficulty
nonce = 0
start = time.time()
while True:
data = f"{block_data}:{nonce}"
hash_result = hashlib.sha256(data.encode()).hexdigest()
print(f"Nonce {nonce}: {hash_result[:10]}...")
if hash_result.startswith(target):
elapsed = time.time() - start
print(f"\nBLOCK MINED!")
print(f"Nonce: {nonce}")
print(f"Hash: {hash_result}")
print(f"Time: {elapsed:.3f}s")
print(f"Hash rate: {nonce/elapsed:.0f} H/s")
return nonce, hash_result
nonce += 1
# Try with difficulty=2 for demonstration
nonce, hash_result = mine_block("Block#1234:Alice->Bob:50BTC", difficulty=2)
Example 2: Build and Verify Merkle Tree
Problem: Given 8 transactions, build a Merkle tree and verify transaction #5 is included.
Solution:
- 8 transactions → 8 leaf hashes
- Pair hashes: (H1+H2), (H3+H4), (H5+H6), (H7+H8) → 4 intermediate
- Pair again: (H12+H34), (H56+H78) → 2 nodes
- Final pair: (H1234+H5678) → Root hash
- Tree depth: log2(8) = 3 levels + root
- Proof for tx[5]: Need sibling hashes at each level (3 hashes)
Example 3: Implement Token Contract
Problem: Create a basic ERC-20 token with mint, burn, and transfer functionality in Python simulation.
Solution:
class Token:
def __init__(self, name, symbol, decimals, initial_supply):
self.name = name
self.symbol = symbol
self.decimals = decimals
self.total_supply = initial_supply
self.balances = {"deployer": initial_supply}
self.allowances = {}
def transfer(self, sender, recipient, amount):
if self.balances.get(sender, 0) < amount:
raise ValueError("Insufficient balance")
self.balances[sender] = self.balances.get(sender, 0) - amount
self.balances[recipient] = self.balances.get(recipient, 0) + amount
return True
def mint(self, to, amount):
self.total_supply += amount
self.balances[to] = self.balances.get(to, 0) + amount
def burn(self, from_address, amount):
if self.balances.get(from_address, 0) < amount:
raise ValueError("Insufficient balance")
self.balances[from_address] -= amount
self.total_supply -= amount
def balance_of(self, address):
return self.balances.get(address, 0)
# Usage
token = Token("EduVerse", "EDU", 18, 1000000)
token.transfer("deployer", "alice", 5000)
token.mint("bob", 1000)
print(f"Alice: {token.balance_of('alice')}") # 5000
print(f"Bob: {token.balance_of('bob')}") # 1000
Example 4: Double-Spend Attack Prevention
Problem: Explain how blockchain prevents double-spending and demonstrate with code.
Solution:
- Problem: Digital money can be copied — how to prevent spending same coin twice?
- Solution: All transactions recorded on shared public ledger
- Process:
- Alice signs transaction: "Alice → Bob: 5 BTC"
- Transaction broadcast to network
- Validators verify Alice has 5 BTC (check ledger history)
- Transaction included in block, block added to chain
- If Alice tries to spend same 5 BTC again → validators reject (balance check fails)
- Key insight: UTXO model (Bitcoin) or account model (Ethereum) tracks all balances
Example 5: Gas Fee Calculation
Problem: Calculate gas fees for Ethereum transactions.
Solution:
Fee = Gas Used × Gas Price (in Gwei)
Gas Used (by operation):
- Simple transfer: 21,000 gas
- ERC-20 transfer: ~65,000 gas
- Smart contract deploy: ~1,000,000+ gas
Example:
Gas Used = 21,000
Gas Price = 20 Gwei = 20 × 10⁻⁹ ETH
Fee = 21,000 × 20 × 10⁻⁹ = 0.00042 ETH
EIP-1559: Base fee (burned) + Priority fee (tip to miner/validator)
Practice Problems
Problem 1: Implement SHA-256 Hash
Write a function that computes the SHA-256 hash of input data and returns the hex digest.
import hashlib
def sha256_hash(data: str) -> str:
"""Compute SHA-256 hash."""
return hashlib.sha256(data.encode()).hexdigest()
# Test
print(sha256_hash("Hello, Blockchain!"))
# 3a7bd3e2360a3d29eea436fcfb7e44c735d117c42d1c1835420b6b9942dd4f1b
Problem 2: Create a Wallet Generator
Generate a blockchain wallet with private key, public key, and address.
import secrets
import hashlib
def generate_wallet():
"""Generate a simple blockchain wallet."""
# Private key: random 256-bit number
private_key = "0x" + secrets.token_hex(32)
# Public key: derived from private key (simplified)
public_key = "0x" + hashlib.sha256(
private_key.encode()
).hexdigest()
# Address: last 40 chars of hash of public key
address = "0x" + hashlib.sha256(
public_key.encode()
).hexdigest()[-40:]
return {
"private_key": private_key,
"public_key": public_key,
"address": address
}
wallet = generate_wallet()
print(f"Address: {wallet['address']}")
print(f"Public Key: {wallet['public_key'][:20]}...")
Problem 3: Verify Blockchain Integrity
Write a function that verifies the entire blockchain hasn't been tampered with.
def verify_blockchain(chain):
"""Verify blockchain integrity."""
for i in range(1, len(chain)):
current = chain[i]
previous = chain[i-1]
# Check hash matches
if current['hash'] != calculate_block_hash(current):
return False, f"Block {i} hash mismatch"
# Check chain linkage
if current['previous_hash'] != previous['hash']:
return False, f"Block {i} chain broken"
# Check proof of work
if not current['hash'].startswith('0000'):
return False, f"Block {i} invalid PoW"
return True, "Blockchain valid"
is_valid, message = verify_blockchain(my_blockchain.chain)
print(message)
Problem 4: Implement UTXO Model
Create a simplified UTXO (Unspent Transaction Output) system like Bitcoin.
class UTXO:
def __init__(self, tx_id, index, amount, recipient):
self.tx_id = tx_id
self.index = index
self.amount = amount
self.recipient = recipient
class UTXOManager:
def __init__(self):
self.utxos = {} # (tx_id, index) -> UTXO
def add_utxo(self, utxo):
self.utxos[(utxo.tx_id, utxo.index)] = utxo
def spend_utxos(self, inputs):
total = 0
for tx_id, index in inputs:
key = (tx_id, index)
if key not in self.utxos:
raise ValueError(f"UTXO {tx_id}:{index} not found")
total += self.utxos[key].amount
del self.utxos[key]
return total
def get_balance(self, address):
return sum(u.amount for u in self.utxos.values()
if u.recipient == address)
Problem 5: Calculate Gas Costs
Estimate gas costs for deploying and interacting with a smart contract.
def estimate_gas(contract_type="ERC20"):
"""Estimate gas costs for common operations."""
costs = {
"ETH_transfer": 21000,
"ERC20_transfer": 65000,
"ERC20_approve": 46000,
"ERC20_deploy": 1_500_000,
"NFT_mint": 80000,
"NFT_transfer": 50000,
"swap": 150000,
}
return costs.get(contract_type, 21000)
def calculate_fee(gas_used, gas_price_gwei):
return gas_used * gas_price_gwei * 1e-9
# Example
deploy_cost = calculate_fee(estimate_gas("ERC20_deploy"), 20)
print(f"Deploy ERC-20: {deploy_cost:.4f} ETH") # 0.03 ETH
Problem 6: Implement Multi-Sig Wallet
Create a multi-signature wallet requiring M-of-N signatures to execute transactions.
class MultiSigWallet:
def __init__(self, owners, required_signatures):
self.owners = set(owners)
self.required = required_signatures
self.transactions = {}
self.tx_count = 0
def submit_transaction(self, tx_id, to, value, data=""):
if tx_id in self.owners:
self.transactions[self.tx_count] = {
"to": to, "value": value, "data": data,
"confirmations": {tx_id}, "executed": False
}
self.tx_count += 1
def confirm_transaction(self, tx_index, confirmer):
if confirmer in self.owners:
self.transactions[tx_index]["confirmations"].add(confirmer)
def execute_transaction(self, tx_index):
tx = self.transactions[tx_index]
if len(tx["confirmations"]) >= self.required:
tx["executed"] = True
return True
return False
Problem 7: Implement Block Explorer
Create a function that displays blockchain information like a block explorer.
def display_block(chain, block_index):
"""Display block information like a block explorer."""
block = chain[block_index]
print(f"=== Block #{block.index} ===")
print(f"Hash: {block.hash}")
print(f"Previous Hash: {block.previous_hash}")
print(f"Timestamp: {time.ctime(block.timestamp)}")
print(f"Nonce: {block.nonce}")
print(f"Transactions: {len(block.transactions)}")
print(f"Merkle Root: {compute_merkle_root(block.transactions)}")
for i, tx in enumerate(block.transactions):
print(f" Tx {i}: {tx.sender[:8]}... → {tx.recipient[:8]}... : {tx.amount}")
Problem 8: Token Swap Calculation
Calculate output amount for a Uniswap-style AMM swap with 0.3% fee.
x × y = k (constant product formula)
Amount In (with fee) = Amount In × 997 / 1000
Amount Out = (y × Amount In With Fee) / (x + Amount In With Fee)
def calculate_swap_output(amount_in, reserve_in, reserve_out, fee=0.003):
amount_in_with_fee = amount_in * (1 - fee)
numerator = amount_in_with_fee * reserve_out
denominator = reserve_in + amount_in_with_fee
return numerator / denominator
# Example: Swap 1 ETH for USDC
# Reserve: 100 ETH = 300,000 USDC
output = calculate_swap_output(1, 100, 300000)
print(f"1 ETH → {output:.2f} USDC") # ~2970 USDC
Problem 9: Implement Voting DApp Backend
Create backend functions for a blockchain voting system.
class VotingSystem:
def __init__(self):
self.proposals = []
self.votes = {} # voter -> proposal_id
self.voters = set()
def register_voter(self, address):
if address not in self.voters:
self.voters.add(address)
return True
return False
def create_proposal(self, name, description):
pid = len(self.proposals)
self.proposals.append({"id": pid, "name": name,
"description": description,
"votes": 0})
return pid
def vote(self, voter, proposal_id):
if voter not in self.voters:
return "Not registered"
if voter in self.votes:
return "Already voted"
self.votes[voter] = proposal_id
self.proposals[proposal_id]["votes"] += 1
return "Vote recorded"
def get_winner(self):
return max(self.proposals, key=lambda p: p["votes"])
Problem 10: Calculate Network Hash Rate
Estimate the network hash rate based on difficulty and block time.
Hash Rate = Difficulty × 2³² / Block Time
Example (Bitcoin):
Difficulty = 50,000,000,000
Block Time = 600 seconds (10 minutes)
Hash Rate = 50×10¹⁰ × 2³² / 600
≈ 358,000 TH/s
≈ 358 EH/s