Chapter 10: Virtual Currency Mining Technology

10.1 Mining Basics

10.1.1 What is Mining

Mining is the process of verifying transactions and creating new blocks through computational problem-solving. It is one of the core security mechanisms of blockchain networks.

import hashlib
import time
import json
import random
from datetime import datetime

class Block:
    def __init__(self, index, transactions, previous_hash, timestamp=None):
        self.index = index
        self.transactions = transactions
        self.previous_hash = previous_hash
        self.timestamp = timestamp or datetime.now()
        self.nonce = 0
        self.hash = None

    def calculate_hash(self):
        """Calculate block hash"""
        block_string = json.dumps({
            'index': self.index,
            'transactions': self.transactions,
            'previous_hash': self.previous_hash,
            'timestamp': self.timestamp.isoformat(),
            'nonce': self.nonce
        }, sort_keys=True)
        return hashlib.sha256(block_string.encode()).hexdigest()

    def mine_block(self, difficulty):
        """Mining process - Proof of Work"""
        target = "0" * difficulty
        start_time = time.time()
        hash_count = 0

        print(f"Starting to mine block {self.index}, difficulty: {difficulty}")
        print(f"Target: hash must start with {target}")

        while True:
            self.hash = self.calculate_hash()
            hash_count += 1

            if self.hash[:difficulty] == target:
                end_time = time.time()
                mining_time = end_time - start_time
                hash_rate = hash_count / mining_time if mining_time > 0 else 0

                print(f"Block mined successfully!")
                print(f"Hash: {self.hash}")
                print(f"Nonce: {self.nonce}")
                print(f"Time taken: {mining_time:.2f}s")
                print(f"Attempts: {hash_count:,}")
                print(f"Hash rate: {hash_rate:,.0f} H/s")
                break

            self.nonce += 1

            # Display progress every 100,000 attempts
            if hash_count % 100000 == 0:
                elapsed = time.time() - start_time
                current_hash_rate = hash_count / elapsed if elapsed > 0 else 0
                print(f"Progress: {hash_count:,} attempts, current hash rate: {current_hash_rate:,.0f} H/s")

class SimpleMiner:
    def __init__(self):
        self.blockchain = []
        self.pending_transactions = []
        self.mining_reward = 10
        self.difficulty = 4

    def create_genesis_block(self):
        """Create genesis block"""
        genesis_block = Block(0, [], "0")
        genesis_block.mine_block(self.difficulty)
        self.blockchain.append(genesis_block)
        return genesis_block

    def get_latest_block(self):
        """Get latest block"""
        return self.blockchain[-1] if self.blockchain else None

    def add_transaction(self, transaction):
        """Add pending transaction"""
        self.pending_transactions.append(transaction)

    def mine_pending_transactions(self, mining_reward_address):
        """Mine and confirm pending transactions"""
        # Add mining reward transaction
        reward_transaction = {
            'from': None,  # None indicates mining reward
            'to': mining_reward_address,
            'amount': self.mining_reward
        }

        self.pending_transactions.append(reward_transaction)

        # Create new block
        previous_block = self.get_latest_block()
        new_block = Block(
            index=len(self.blockchain),
            transactions=self.pending_transactions.copy(),
            previous_hash=previous_block.hash if previous_block else "0"
        )

        # Mine
        new_block.mine_block(self.difficulty)

        # Add to blockchain
        self.blockchain.append(new_block)

        # Clear pending transactions
        self.pending_transactions = []

        return new_block

    def adjust_difficulty(self, target_time=10):
        """Adjust mining difficulty"""
        if len(self.blockchain) < 2:
            return

        latest_block = self.blockchain[-1]
        previous_block = self.blockchain[-2]

        # Calculate actual block time
        actual_time = (latest_block.timestamp - previous_block.timestamp).total_seconds()

        # Adjust difficulty
        if actual_time < target_time / 2:
            self.difficulty += 1
            print(f"Mining too fast, difficulty increased to: {self.difficulty}")
        elif actual_time > target_time * 2:
            self.difficulty = max(1, self.difficulty - 1)
            print(f"Mining too slow, difficulty decreased to: {self.difficulty}")

# Mining demonstration
print("Virtual Currency Mining Demonstration")
print("=" * 50)

miner = SimpleMiner()

# Create genesis block
print("Creating genesis block...")
genesis = miner.create_genesis_block()

# Add some transactions
transactions = [
    {'from': 'Alice', 'to': 'Bob', 'amount': 50},
    {'from': 'Bob', 'to': 'Charlie', 'amount': 25},
    {'from': 'Charlie', 'to': 'Alice', 'amount': 10}
]

for tx in transactions:
    miner.add_transaction(tx)

print(f"\nPending transactions: {len(miner.pending_transactions)}")

# Mine
print("\nStarting mining...")
miner_address = "MinerRewardAddress123"
new_block = miner.mine_pending_transactions(miner_address)

print(f"\nBlockchain length: {len(miner.blockchain)}")
print(f"Latest block contains {len(new_block.transactions)} transactions")

# Display blockchain info
print("\nBlockchain Info:")
for block in miner.blockchain:
    print(f"Block {block.index}: {block.hash[:10]}... (Transactions: {len(block.transactions)})")

10.1.2 Mining Difficulty Adjustment Mechanism

import matplotlib.pyplot as plt
import numpy as np

class DifficultyAdjustment:
    def __init__(self, target_time=600, adjustment_interval=2016):  # Bitcoin parameters
        self.target_time = target_time  # Target block time (seconds)
        self.adjustment_interval = adjustment_interval  # Adjustment interval (blocks)
        self.difficulty_history = []
        self.block_times = []

    def calculate_new_difficulty(self, actual_time_span, current_difficulty):
        """Calculate new difficulty"""
        expected_time_span = self.target_time * self.adjustment_interval

        # Limit adjustment range (Bitcoin max 4x adjustment)
        if actual_time_span < expected_time_span / 4:
            actual_time_span = expected_time_span / 4
        if actual_time_span > expected_time_span * 4:
            actual_time_span = expected_time_span * 4

        new_difficulty = current_difficulty * expected_time_span / actual_time_span
        return new_difficulty

    def simulate_mining(self, num_blocks=10000, hash_rate_changes=None):
        """Simulate mining process"""
        current_difficulty = 1.0
        current_hash_rate = 1000  # TH/s

        for block_num in range(num_blocks):
            # Simulate hash rate changes
            if hash_rate_changes and block_num in hash_rate_changes:
                current_hash_rate *= hash_rate_changes[block_num]
                print(f"Block {block_num}: Hash rate changed to {current_hash_rate:.0f} TH/s")

            # Calculate block time (simplified model)
            expected_time = self.target_time * current_difficulty / current_hash_rate
            # Add randomness
            actual_time = np.random.exponential(expected_time)

            self.block_times.append(actual_time)

            # Difficulty adjustment
            if (block_num + 1) % self.adjustment_interval == 0:
                total_time = sum(self.block_times[-self.adjustment_interval:])
                new_difficulty = self.calculate_new_difficulty(total_time, current_difficulty)

                print(f"Block {block_num + 1}: Difficulty adjustment")
                print(f"  Actual time: {total_time:.0f}s ({total_time/3600:.1f}h)")
                print(f"  Expected time: {self.target_time * self.adjustment_interval:.0f}s")
                print(f"  Difficulty: {current_difficulty:.6f} -> {new_difficulty:.6f}")
                print(f"  Adjustment factor: {new_difficulty/current_difficulty:.2f}x")

                current_difficulty = new_difficulty

            self.difficulty_history.append(current_difficulty)

        return self.difficulty_history, self.block_times

    def analyze_stability(self):
        """Analyze mining stability"""
        if len(self.block_times) < 100:
            return {}

        recent_times = self.block_times[-100:]  # Last 100 blocks
        avg_time = np.mean(recent_times)
        std_time = np.std(recent_times)
        target_deviation = abs(avg_time - self.target_time) / self.target_time

        return {
            'average_block_time': avg_time,
            'target_block_time': self.target_time,
            'deviation_from_target': target_deviation,
            'standard_deviation': std_time,
            'stability_score': 1 / (1 + target_deviation)  # 0-1 score
        }

# Difficulty adjustment simulation
print("\nMining Difficulty Adjustment Simulation")
print("=" * 40)

difficulty_system = DifficultyAdjustment(target_time=600, adjustment_interval=100)  # Simplified parameters

# Simulate hash rate change scenarios
hash_rate_changes = {
    2000: 2.0,  # Hash rate doubles at block 2000
    4000: 0.5,  # Hash rate halves at block 4000
    6000: 3.0,  # Hash rate triples at block 6000
    8000: 0.3   # Hash rate drops significantly at block 8000
}

difficulties, block_times = difficulty_system.simulate_mining(
    num_blocks=10000,
    hash_rate_changes=hash_rate_changes
)

# Analyze stability
stability = difficulty_system.analyze_stability()
print(f"\nMining Stability Analysis:")
print(f"Average block time: {stability['average_block_time']:.0f}s")
print(f"Target block time: {stability['target_block_time']:.0f}s")
print(f"Deviation: {stability['deviation_from_target']:.2%}")
print(f"Stability score: {stability['stability_score']:.3f}")

10.2 Detailed Consensus Algorithms

10.2.1 Proof of Work (PoW)

import random
from concurrent.futures import ThreadPoolExecutor
import threading

class ProofOfWork:
    def __init__(self, difficulty=4):
        self.difficulty = difficulty
        self.target = 2 ** (256 - difficulty * 4)

    def validate_proof(self, block_data, nonce):
        """Validate proof of work"""
        hash_result = hashlib.sha256(f"{block_data}{nonce}".encode()).hexdigest()
        return int(hash_result, 16) < self.target

    def mine_block(self, block_data, start_nonce=0, max_nonce=None):
        """Mining process"""
        nonce = start_nonce
        start_time = time.time()

        while max_nonce is None or nonce < max_nonce:
            if self.validate_proof(block_data, nonce):
                end_time = time.time()
                return {
                    'success': True,
                    'nonce': nonce,
                    'hash': hashlib.sha256(f"{block_data}{nonce}".encode()).hexdigest(),
                    'time_taken': end_time - start_time,
                    'attempts': nonce - start_nonce + 1
                }
            nonce += 1

        return {'success': False, 'attempts': nonce - start_nonce}

class MiningPool:
    def __init__(self, difficulty=4, num_miners=4):
        self.pow = ProofOfWork(difficulty)
        self.num_miners = num_miners
        self.mining_results = []
        self.mining_active = False
        self.winner_found = threading.Event()

    def miner_worker(self, miner_id, block_data, nonce_range):
        """Miner worker function"""
        start_nonce, end_nonce = nonce_range
        print(f"Miner {miner_id} started mining: nonce range {start_nonce:,} - {end_nonce:,}")

        current_nonce = start_nonce
        start_time = time.time()

        while current_nonce < end_nonce and not self.winner_found.is_set():
            if self.pow.validate_proof(block_data, current_nonce):
                self.winner_found.set()
                end_time = time.time()

                result = {
                    'miner_id': miner_id,
                    'success': True,
                    'nonce': current_nonce,
                    'hash': hashlib.sha256(f"{block_data}{current_nonce}".encode()).hexdigest(),
                    'time_taken': end_time - start_time,
                    'attempts': current_nonce - start_nonce + 1
                }

                self.mining_results.append(result)
                print(f"🎉 Miner {miner_id} found solution!")
                print(f"Nonce: {current_nonce}, Hash: {result['hash'][:16]}...")
                return result

            current_nonce += 1

            # Check every 10,000 attempts if another miner found solution
            if current_nonce % 10000 == 0 and self.winner_found.is_set():
                break

        # If no solution found
        if not self.winner_found.is_set():
            end_time = time.time()
            result = {
                'miner_id': miner_id,
                'success': False,
                'time_taken': end_time - start_time,
                'attempts': current_nonce - start_nonce
            }
            self.mining_results.append(result)

        return None

    def start_mining(self, block_data):
        """Start distributed mining"""
        self.mining_results = []
        self.winner_found.clear()

        # Allocate nonce range for each miner
        range_size = 2 ** 20  # 1 million nonces per miner
        nonce_ranges = []
        for i in range(self.num_miners):
            start = i * range_size
            end = start + range_size
            nonce_ranges.append((start, end))

        print(f"Starting parallel mining with {self.num_miners} miners...")
        print(f"Target: hash must start with {self.pow.difficulty} zeros")

        # Use thread pool for parallel mining
        with ThreadPoolExecutor(max_workers=self.num_miners) as executor:
            futures = []
            for i, nonce_range in enumerate(nonce_ranges):
                future = executor.submit(self.miner_worker, i+1, block_data, nonce_range)
                futures.append(future)

            # Wait for first successful result
            for future in futures:
                try:
                    result = future.result(timeout=10)  # 10 second timeout
                    if result and result['success']:
                        break
                except:
                    continue

        return self.mining_results

    def get_mining_statistics(self):
        """Get mining statistics"""
        if not self.mining_results:
            return {}

        total_attempts = sum(result['attempts'] for result in self.mining_results)
        total_time = max(result['time_taken'] for result in self.mining_results)
        winner = next((r for r in self.mining_results if r.get('success')), None)

        stats = {
            'total_miners': len(self.mining_results),
            'total_attempts': total_attempts,
            'total_time': total_time,
            'combined_hash_rate': total_attempts / total_time if total_time > 0 else 0,
            'winner': winner
        }

        return stats

# PoW mining demonstration
print("\nProof of Work Mining Demonstration")
print("=" * 40)

# Create block data
block_data = {
    'index': 1,
    'transactions': ['Alice->Bob: 10 BTC', 'Bob->Charlie: 5 BTC'],
    'previous_hash': '000abc123def456...',
    'timestamp': datetime.now().isoformat()
}

block_string = json.dumps(block_data, sort_keys=True)

# Single miner mining
print("Single miner mining:")
pow = ProofOfWork(difficulty=4)
result = pow.mine_block(block_string)

if result['success']:
    print(f"Mining successful!")
    print(f"Nonce: {result['nonce']}")
    print(f"Hash: {result['hash']}")
    print(f"Time: {result['time_taken']:.2f}s")
    print(f"Attempts: {result['attempts']:,}")
    hash_rate = result['attempts'] / result['time_taken'] if result['time_taken'] > 0 else 0
    print(f"Hash rate: {hash_rate:,.0f} H/s")

# Multi-miner competitive mining
print(f"\nMulti-miner competitive mining:")
mining_pool = MiningPool(difficulty=4, num_miners=4)
pool_results = mining_pool.start_mining(block_string)

stats = mining_pool.get_mining_statistics()
if stats:
    print(f"\nMining Statistics:")
    print(f"Participating miners: {stats['total_miners']}")
    print(f"Total attempts: {stats['total_attempts']:,}")
    print(f"Total time: {stats['total_time']:.2f}s")
    print(f"Combined hash rate: {stats['combined_hash_rate']:,.0f} H/s")

    if stats['winner']:
        winner = stats['winner']
        print(f"Winning miner: Miner {winner['miner_id']}")
        print(f"Winning time: {winner['time_taken']:.2f}s")

10.2.2 Proof of Stake (PoS)

class ProofOfStake:
    def __init__(self):
        self.validators = {}
        self.total_stake = 0
        self.validator_history = {}
        self.slashing_conditions = []

    def add_validator(self, validator_id, stake_amount):
        """Add validator"""
        if validator_id in self.validators:
            self.validators[validator_id]['stake'] += stake_amount
        else:
            self.validators[validator_id] = {
                'stake': stake_amount,
                'blocks_validated': 0,
                'rewards_earned': 0,
                'slashed': False,
                'join_time': datetime.now()
            }

        self.total_stake += stake_amount
        self._update_validator_weights()

    def _update_validator_weights(self):
        """Update validator weights"""
        if self.total_stake > 0:
            for validator_id in self.validators:
                stake = self.validators[validator_id]['stake']
                self.validators[validator_id]['weight'] = stake / self.total_stake

    def select_validator(self, block_height, randomness_seed=None):
        """Select next block validator"""
        if not self.validators or self.total_stake == 0:
            return None

        # Simplified version of Verifiable Random Function (VRF)
        if randomness_seed is None:
            randomness_seed = block_height

        random.seed(randomness_seed)

        # Weighted random selection
        validators_list = []
        weights = []

        for validator_id, validator_data in self.validators.items():
            if not validator_data['slashed']:  # Slashed validators cannot participate
                validators_list.append(validator_id)
                weights.append(validator_data['weight'])

        if not validators_list:
            return None

        selected_validator = random.choices(validators_list, weights=weights)[0]
        return selected_validator

    def validate_block(self, validator_id, block_data):
        """Validator validates block"""
        if validator_id not in self.validators:
            return {'success': False, 'error': 'Validator not found'}

        validator = self.validators[validator_id]
        if validator['slashed']:
            return {'success': False, 'error': 'Validator is slashed'}

        # Simulate validation process
        validation_time = time.time()

        # Basic validation checks
        validation_checks = {
            'format_valid': self._check_block_format(block_data),
            'transactions_valid': self._check_transactions(block_data),
            'previous_hash_valid': self._check_previous_hash(block_data)
        }

        all_valid = all(validation_checks.values())

        if all_valid:
            # Validation successful, give reward
            reward = self._calculate_reward(validator_id)
            validator['blocks_validated'] += 1
            validator['rewards_earned'] += reward

            result = {
                'success': True,
                'validator_id': validator_id,
                'validation_time': validation_time,
                'reward': reward,
                'checks': validation_checks
            }
        else:
            # Validation failed, possible penalty
            penalty = self._calculate_penalty(validator_id, validation_checks)

            result = {
                'success': False,
                'validator_id': validator_id,
                'validation_time': validation_time,
                'penalty': penalty,
                'checks': validation_checks
            }

        return result

    def _check_block_format(self, block_data):
        """Check block format"""
        required_fields = ['index', 'transactions', 'previous_hash', 'timestamp']
        return all(field in block_data for field in required_fields)

    def _check_transactions(self, block_data):
        """Check transaction validity"""
        # Simplified check
        transactions = block_data.get('transactions', [])
        return len(transactions) <= 1000  # Maximum transaction limit

    def _check_previous_hash(self, block_data):
        """Check previous block hash"""
        # Simplified check
        prev_hash = block_data.get('previous_hash', '')
        return len(prev_hash) == 64  # SHA256 hash length

    def _calculate_reward(self, validator_id):
        """Calculate validation reward"""
        base_reward = 10
        stake_bonus = self.validators[validator_id]['stake'] * 0.001
        return base_reward + stake_bonus

    def _calculate_penalty(self, validator_id, validation_checks):
        """Calculate penalty"""
        failed_checks = sum(1 for check in validation_checks.values() if not check)
        penalty_rate = failed_checks * 0.1  # 10% penalty per failed check

        validator = self.validators[validator_id]
        penalty = validator['stake'] * penalty_rate

        # Apply penalty
        if penalty > 0:
            validator['stake'] = max(0, validator['stake'] - penalty)
            self.total_stake = max(0, self.total_stake - penalty)
            self._update_validator_weights()

        return penalty

    def implement_slashing(self, validator_id, reason):
        """Implement slashing"""
        if validator_id not in self.validators:
            return False

        validator = self.validators[validator_id]
        slashed_amount = validator['stake']

        # Confiscate all stake
        validator['stake'] = 0
        validator['slashed'] = True
        validator['slashing_reason'] = reason
        validator['slashing_time'] = datetime.now()

        self.total_stake -= slashed_amount
        self._update_validator_weights()

        print(f"Validator {validator_id} slashed, reason: {reason}")
        print(f"Slashed amount: {slashed_amount}")

        return True

    def get_validator_info(self, validator_id):
        """Get validator information"""
        if validator_id not in self.validators:
            return None

        validator = self.validators[validator_id]
        return {
            'validator_id': validator_id,
            'stake': validator['stake'],
            'weight': validator.get('weight', 0),
            'blocks_validated': validator['blocks_validated'],
            'rewards_earned': validator['rewards_earned'],
            'slashed': validator['slashed'],
            'join_time': validator['join_time']
        }

    def get_network_stats(self):
        """Get network statistics"""
        active_validators = sum(1 for v in self.validators.values() if not v['slashed'])
        total_blocks_validated = sum(v['blocks_validated'] for v in self.validators.values())

        return {
            'total_validators': len(self.validators),
            'active_validators': active_validators,
            'total_stake': self.total_stake,
            'total_blocks_validated': total_blocks_validated,
            'average_stake': self.total_stake / len(self.validators) if self.validators else 0
        }

# PoS demonstration
print("\nProof of Stake Mining Demonstration")
print("=" * 40)

pos_network = ProofOfStake()

# Add validators
validators_data = [
    ('Validator_A', 1000),
    ('Validator_B', 2000),
    ('Validator_C', 500),
    ('Validator_D', 1500)
]

print("Adding validators:")
for validator_id, stake in validators_data:
    pos_network.add_validator(validator_id, stake)
    info = pos_network.get_validator_info(validator_id)
    print(f"{validator_id}: Stake {stake}, Weight {info['weight']:.2%}")

# Simulate block validation process
print(f"\nSimulating block validation:")
for block_height in range(1, 6):
    # Select validator
    selected_validator = pos_network.select_validator(block_height)

    if selected_validator:
        print(f"\nBlock {block_height}:")
        print(f"Selected validator: {selected_validator}")

        # Create test block
        test_block = {
            'index': block_height,
            'transactions': [f'tx_{i}' for i in range(random.randint(1, 10))],
            'previous_hash': 'a' * 64,  # Simulated hash
            'timestamp': datetime.now().isoformat()
        }

        # Validate block
        validation_result = pos_network.validate_block(selected_validator, test_block)

        if validation_result['success']:
            print(f"✅ Validation successful, reward: {validation_result['reward']:.2f}")
        else:
            print(f"❌ Validation failed, penalty: {validation_result.get('penalty', 0):.2f}")

# Display validator status
print(f"\nFinal validator status:")
for validator_id in pos_network.validators:
    info = pos_network.get_validator_info(validator_id)
    print(f"{validator_id}:")
    print(f"  Stake: {info['stake']:.2f}")
    print(f"  Blocks validated: {info['blocks_validated']}")
    print(f"  Rewards earned: {info['rewards_earned']:.2f}")

# Network statistics
stats = pos_network.get_network_stats()
print(f"\nNetwork statistics:")
print(f"Total validators: {stats['total_validators']}")
print(f"Active validators: {stats['active_validators']}")
print(f"Total stake: {stats['total_stake']:.2f}")
print(f"Total blocks validated: {stats['total_blocks_validated']}")

(Continuing with remaining sections on mining hardware, environmental sustainability, etc…)

10.5 Course Summary

This chapter detailed various aspects of virtual currency mining technology:

Key Points

1. Mining Principles: Proof of Work is the foundation of Bitcoin’s security
2. Consensus Mechanisms: PoW and PoS each have advantages and disadvantages, suitable for different scenarios
3. Hardware Evolution: ASIC miners provide optimal efficiency and returns
4. Mining Pool Model: Diversifies risk, stabilizes returns
5. Environmental Challenges: Sustainable development is an important direction for industry growth

Technology Development Trends

• More efficient mining hardware
• Widespread adoption of green energy
• Achievement of carbon-neutral mining
• Intelligent mining farm management

Investment Recommendations

• Focus on energy costs and policy environment
• Choose high-efficiency mining equipment
• Consider joining reputable mining pools
• Prioritize environmental compliance requirements

The development of mining technology reflects the maturation of the entire cryptocurrency ecosystem. As environmental requirements become increasingly stringent and technology continues to advance, sustainable mining will become the core competitiveness of the industry. Through this chapter, you should understand the technical principles, economic models, and development directions of mining.