比特幣區塊鏈數據分析 Python 實作教學：從基礎到進階
完整介紹如何使用 Python 進行比特幣區塊鏈數據分析，涵蓋 Bitcoin Core RPC、公共 API、鏈上指標計算、UTXO 分析等實作範例。
比特幣區塊鏈數據分析 Python 實作教學：從基礎到進階
比特幣區塊鏈蘊含豐富的數據資源，這些數據可以用於市場分析、風險評估、投資決策支援等多種應用場景。本文提供完整的 Python 實作教學，從基礎的區塊數據讀取到進階的鏈上指標計算，幫助讀者建立比特幣數據分析的實作能力。所有程式碼均可直接執行，並提供詳細的註解說明。
比特幣區塊鏈數據概述
比特幣區塊結構詳解
比特幣區塊由區塊頭（Block Header）和區塊體（Block Body）兩部分組成。理解區塊結構是進行鏈上數據分析的基礎。
比特幣區塊結構：
┌─────────────────────────────────────────────────────┐
│ 區塊頭（80 bytes） │
├─────────────────────────────────────────────────────┤
│ 版本號 (4 bytes) │ 4 字節 │
│ 父區塊哈希 (32 bytes) │ 32 字節 │
│ Merkle 根 (32 bytes) │ 32 字節 │
│ 時間戳 (4 bytes) │ 4 字節 │
│ 難度目標 (4 bytes) │ 4 字節 │
│ 隨機數 Nonce (4 bytes) │ 4 字節 │
├─────────────────────────────────────────────────────┤
│ 交易計數器 │
├─────────────────────────────────────────────────────┤
│ 交易列表 │
│ (交易數量可變，每筆交易至少 125 bytes) │
└─────────────────────────────────────────────────────┘
區塊大小限制：
├── 區塊上限：4,000,000 weight units (WU)
├── 原始資料上限：1,000,000 bytes
├── 隔離見證升級後：區塊可容納更多交易
└── 平均區塊大小：1.5-3.5 MB比特幣網路關鍵指標數據
比特幣網路的運作產生了多維度的數據指標，這些指標反映了網路的健康狀態、使用情況和經濟活動。
比特幣網路關鍵指標數據類別：
1. 區塊數據
├── 區塊高度 (Block Height)
├── 區塊大小 (Block Size)
├── 區塊重量 (Block Weight)
├── 交易數量 (Transaction Count)
└── 區塊時間戳 (Timestamp)
2. 交易數據
├── 交易 ID (TXID)
├── 輸入數量 (Input Count)
├── 輸出數量 (Output Count)
├── 交易大小 (Transaction Size)
├── 手续费 (Fee)
└── 費率 (Fee Rate: sat/vB)
3. UTXO 數據
├── 未花費輸出總數
├── UTXO 集合大小
├── 總 BTC 鎖定價值
└── 平均 UTXO 年齡
4. 網路數據
├── 網路算力 (Hash Rate)
├── 難度 (Difficulty)
├── 出塊時間 (Block Time)
├── 節點數量 (Node Count)
└── 交易池大小 (Mempool Size)比特幣數據獲取 API 詳解
使用 Bitcoin Core RPC 獲取數據
Bitcoin Core 提供了完整的 JSON-RPC API，可以用於獲取各種類型的區塊鏈數據。
#!/usr/bin/env python3
"""
比特幣區塊鏈數據獲取模組
使用 Bitcoin Core RPC API
"""
import requests
import json
from typing import Dict, List, Optional, Any
from dataclasses import dataclass
from datetime import datetime
import time
class BitcoinRPCClient:
"""比特幣 Core RPC 客戶端"""
def __init__(self, rpc_user: str = "bitcoin",
rpc_password: str = "password",
rpc_host: str = "127.0.0.1",
rpc_port: int = 8332,
use_https: bool = False):
"""
初始化 RPC 客戶端
Args:
rpc_user: RPC 用戶名
rpc_password: RPC 密碼
rpc_host: RPC 主機地址
rpc_port: RPC 端口（8332 主網，18332 測試網）
use_https: 是否使用 HTTPS
"""
self.rpc_user = rpc_user
self.rpc_password = rpc_password
self.rpc_host = rpc_host
self.rpc_port = rpc_port
protocol = "https" if use_https else "http"
self.url = f"{protocol}://{rpc_host}:{rpc_port}"
self.session = requests.Session()
self.session.auth = (rpc_user, rpc_password)
self.session.headers.update({'Content-Type': 'application/json'})
def call(self, method: str, *params) -> Any:
"""
發送 RPC 請求
Args:
method: RPC 方法名稱
params: 方法參數
Returns:
RPC 響應結果
"""
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": 1
}
response = self.session.post(self.url, json=payload, timeout=30)
response.raise_for_status()
result = response.json()
if "error" in result and result["error"] is not None:
raise RuntimeError(f"RPC Error: {result['error']}")
return result["result"]
# ============ 區塊相關方法 ============
def get_block_hash(self, height: int) -> str:
"""獲取指定高度的區塊哈希"""
return self.call("getblockhash", height)
def get_block(self, block_hash: str, verbosity: int = 1) -> Dict:
"""
獲取區塊數據
Args:
block_hash: 區塊哈希
verbosity: 0=原始數據, 1=交易哈希, 2=完整交易數據
Returns:
區塊數據字典
"""
return self.call("getblock", block_hash, verbosity)
def get_block_by_height(self, height: int, verbosity: int = 1) -> Dict:
"""通過高度獲取區塊數據"""
block_hash = self.get_block_hash(height)
return self.get_block(block_hash, verbosity)
def get_best_block_hash(self) -> str:
"""獲取最新區塊哈希"""
return self.call("getbestblockhash")
def get_block_count(self) -> int:
"""獲取當前區塊高度"""
return self.call("getblockcount")
# ============ 交易相關方法 ============
def get_raw_transaction(self, txid: str, verbose: bool = True) -> Dict:
"""
獲取原始交易數據
Args:
txid: 交易 ID
verbose: 是否返回詳細信息
Returns:
交易數據字典
"""
return self.call("getrawtransaction", txid, verbose)
def decode_raw_transaction(self, hex_string: str) -> Dict:
"""解碼原始交易"""
return self.call("decoderawtransaction", hex_string)
# ============ 鏈上指標相關方法 ============
def get_network_hash_rate(self, blocks: int = 120) -> float:
"""
獲取網路算力（過去 N 區塊的平均值）
Args:
blocks: 計算區塊數（預設 120）
Returns:
算力（Hash/s）
"""
return self.call("getnetworkhashps", blocks)
def get_difficulty(self) -> float:
"""獲取當前挖礦難度"""
return self.call("getdifficulty")
# =========內存池相關方法=========
def get_raw_mempool(self, verbose: bool = False) -> List:
"""獲取內存池中的所有交易"""
return self.call("getrawmempool", verbose)
def get_mempool_entry(self, txid: str) -> Dict:
"""獲取內存池中特定交易的詳細信息"""
return self.call("getmempoolentry", txid)
# ========= UTXO 相關方法 ============
def get_utxo_set(self, include_mempool: bool = True) -> Dict:
"""
獲取 UTXO 集合摘要
Args:
include_mempool: 是否包含內存池中的 UTXO
Returns:
UTXO 集合信息
"""
return self.call("gettxoutsetinfo", include_mempool)
# 使用範例
if __name__ == "__main__":
# 連接到本地 Bitcoin Core 節點
# 注意：需要先啟動 bitcoind 並啟用 RPC
# bitcoind -server=1 -rpcuser=user -rpcpassword=password
try:
client = BitcoinRPCClient(
rpc_user="bitcoin",
rpc_password="your_password"
)
# 獲取最新區塊高度
block_count = client.get_block_count()
print(f"當前區塊高度: {block_count:,}")
# 獲取最新區塊
latest_hash = client.get_best_block_hash()
latest_block = client.get_block(latest_hash, verbosity=1)
print(f"\n最新區塊信息：")
print(f" 哈希: {latest_hash[:16]}...")
print(f" 時間: {datetime.fromtimestamp(latest_block['time'])}")
print(f" 交易數: {len(latest_block['tx'])}")
print(f" 大小: {latest_block['size']:,} bytes")
print(f" 重量: {latest_block['weight']:,} WU")
# 獲取網路算力
hash_rate = client.get_network_hash_rate()
print(f"\n網路算力: {hash_rate/1e18:.2f} EH/s")
# 獲取難度
difficulty = client.get_difficulty()
print(f"挖礦難度: {difficulty/1e12:.2f} T")
except Exception as e:
print(f"連接錯誤: {e}")
print("請確保 Bitcoin Core 節點正在運行並啟用 RPC")使用公共 API 獲取比特幣數據
除了 Bitcoin Core RPC，還可以使用多個公共 API 服務獲取比特幣數據，無需運行自己的節點。
#!/usr/bin/env python3
"""
比特幣區塊鏈數據獲取模組
使用公共 API 服務（無需運行節點）
"""
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import time
class BlockchainDataProvider:
"""區塊鏈數據提供商"""
def __init__(self, provider: str = "blockchain.info"):
"""
初始化數據提供商
Args:
provider: 數據提供商名稱
可選: "blockchain.info", "blockstream", "mempool.space"
"""
self.provider = provider
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'BitcoinDataAnalysis/1.0'
})
# ============ Blockchain.info API ============
def get_blockchain_info_blockchain_info(self) -> Dict:
"""獲取區塊鏈基本信息（Blockchain.info）"""
url = "https://blockchain.info/stats?format=json"
response = self.session.get(url, timeout=30)
return response.json()
def get_block_data_blockchain_info(self, block_height: int) -> Dict:
"""獲取指定高度的區塊數據（Blockchain.info）"""
url = f"https://blockchain.info/block-height/{block_height}?format=json"
response = self.session.get(url, timeout=30)
data = response.json()
# 返回最後一個區塊（因為同一高度可能有多個區塊）
return data['blocks'][-1] if data['blocks'] else None
def get_address_info_blockchain_info(self, address: str) -> Dict:
"""獲取地址信息（Blockchain.info）"""
url = f"https://blockchain.info/rawaddr/{address}?limit=50"
response = self.session.get(url, timeout=30)
return response.json()
def get_transaction_blockchain_info(self, tx_hash: str) -> Dict:
"""獲取交易詳情（Blockchain.info）"""
url = f"https://blockchain.info/rawtx/{tx_hash}?format=json"
response = self.session.get(url, timeout=30)
return response.json()
# ============ Blockstream API ============
def get_block_hash_blockstream(self, height: int) -> str:
"""獲取指定高度的區塊哈希（Blockstream）"""
url = f"https://blockstream.info/api/block-height/{height}"
response = self.session.get(url, timeout=30)
return response.text.strip()
def get_block_blockstream(self, block_hash: str) -> Dict:
"""獲取區塊數據（Blockstream）"""
url = f"https://blockstream.info/api/block/{block_hash}"
response = self.session.get(url, timeout=30)
return response.json()
def get_blockTransactions_blockstream(self, block_hash: str, start_index: int = 0) -> Dict:
"""獲取區塊交易列表（Blockstream）"""
url = f"https://blockstream.info/api/block/{block_hash}/txs/{start_index}"
response = self.session.get(url, timeout=30)
return response.json()
def get_address_transactions_blockstream(self, address: str,
last_seen_txid: Optional[str] = None) -> List[Dict]:
"""獲取地址交易歷史（Blockstream）"""
if last_seen_txid:
url = f"https://blockstream.info/api/address/{address}/txs/{last_seen_txid}"
else:
url = f"https://blockstream.info/api/address/{address}/txs"
response = self.session.get(url, timeout=30)
return response.json()
# ============ Mempool.space API ============
def get_mempool_mempool_space(self) -> List[Dict]:
"""獲取內存池交易（Mempool.space）"""
url = "https://mempool.space/api/mempool"
response = self.session.get(url, timeout=30)
return response.json()
def get_fee_recommendations_mempool_space(self) -> Dict:
"""獲取費用建議（Mempool.space）"""
url = "https://mempool.space/api/v1/fees/recommended"
response = self.session.get(url, timeout=30)
return response.json()
def get_blocks_mempool_space(self, height: int) -> List[Dict]:
"""獲取區塊列表（Mempool.space）"""
url = f"https://mempool.space/api/blocks/{height}"
response = self.session.get(url, timeout=30)
return response.json()
# ============ 統一接口 ============
def get_latest_block_height(self) -> int:
"""獲取最新區塊高度"""
if self.provider == "blockchain.info":
info = self.get_blockchain_info_blockchain_info()
return info['n_blocks_total']
elif self.provider == "blockstream":
url = "https://blockstream.info/api/blocks/tip/height"
response = self.session.get(url, timeout=30)
return int(response.text.strip())
elif self.provider == "mempool.space":
url = "https://mempool.space/api/blocks/tip/height"
response = self.session.get(url, timeout=30)
return int(response.text.strip())
else:
raise ValueError(f"Unknown provider: {self.provider}")
def get_address_info(self, address: str) -> Dict:
"""統一獲取地址信息"""
if self.provider == "blockchain.info":
return self.get_address_info_blockchain_info(address)
elif self.provider == "blockstream":
url = f"https://blockstream.info/api/address/{address}"
response = self.session.get(url, timeout=30)
return response.json()
else:
raise ValueError(f"Provider {self.provider} not supported for address info")
# 使用範例
if __name__ == "__main__":
# 使用 Blockchain.info
provider = BlockchainDataProvider("blockchain.info")
try:
# 獲取最新區塊高度
latest_height = provider.get_latest_block_height()
print(f"最新區塊高度: {latest_height:,}")
# 獲取區塊鏈信息
info = provider.get_blockchain_info_blockchain_info()
print(f"\n區塊鏈統計信息：")
print(f" 交易總數: {info['total_transactions_total']:,}")
print(f" 錢包用戶: {info['totalbc_balance']/1e8:.2f} BTC")
print(f" 網路算力: {info['hash_rate']/1e12:.2f} TH/s")
# 測試 Blockstream API
provider2 = BlockchainDataProvider("blockstream")
latest_height2 = provider2.get_latest_block_height()
print(f"\nBlockstream 最新區塊高度: {latest_height2:,}")
except requests.exceptions.RequestException as e:
print(f"API 請求錯誤: {e}")比特幣鏈上指標計算實作
基礎指標計算
以下程式碼展示如何計算比特幣區塊鏈的基礎指標。
#!/usr/bin/env python3
"""
比特幣鏈上指標計算模組
"""
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
from dataclasses import dataclass
@dataclass
class BlockData:
"""區塊數據類"""
height: int
hash: str
timestamp: int
size: int
weight: int
transaction_count: int
fees: int
miner_reward: int
class OnChainMetricsCalculator:
"""比特幣鏈上指標計算器"""
def __init__(self, rpc_client=None):
self.rpc_client = rpc_client
self.cache = {}
def fetch_recent_blocks(self, num_blocks: int = 100) -> List[BlockData]:
"""獲取最近 N 個區塊的數據"""
if not self.rpc_client:
raise ValueError("需要 Bitcoin RPC 客戶端")
blocks = []
current_height = self.rpc_client.get_block_count()
for i in range(num_blocks):
try:
block_hash = self.rpc_client.get_block_hash(current_height - i)
block = self.rpc_client.get_block(block_hash, verbosity=1)
# 計算區塊總費用
# 這裡需要完整交易數據
fees = 0 # 簡化處理
block_data = BlockData(
height=block['height'],
hash=block['hash'],
timestamp=block['time'],
size=block['size'],
weight=block['weight'],
transaction_count=len(block['tx']),
fees=0,
miner_reward=0
)
blocks.append(block_data)
except Exception as e:
print(f"獲取區塊 {current_height - i} 失敗: {e}")
continue
return blocks
def calculate_block_size_stats(self, blocks: List[BlockData]) -> Dict:
"""計算區塊大小統計"""
sizes = [b.size for b in blocks]
return {
"mean": np.mean(sizes),
"median": np.median(sizes),
"std": np.std(sizes),
"min": np.min(sizes),
"max": np.max(sizes),
"p25": np.percentile(sizes, 25),
"p75": np.percentile(sizes, 75),
"p95": np.percentile(sizes, 95),
"p99": np.percentile(sizes, 99)
}
def calculate_transaction_count_stats(self, blocks: List[BlockData]) -> Dict:
"""計算每區塊交易數統計"""
tx_counts = [b.transaction_count for b in blocks]
return {
"mean": np.mean(tx_counts),
"median": np.median(tx_counts),
"std": np.std(tx_counts),
"min": np.min(tx_counts),
"max": np.max(tx_counts),
"total": sum(tx_counts)
}
def calculate_block_time_stats(self, blocks: List[BlockData]) -> Dict:
"""計算區塊時間統計"""
timestamps = sorted([b.timestamp for b in blocks], reverse=True)
time_diffs = [timestamps[i] - timestamps[i+1] for i in range(len(timestamps)-1)]
return {
"mean_seconds": np.mean(time_diffs),
"median_seconds": np.median(time_diffs),
"std_seconds": np.std(time_diffs),
"min_seconds": np.min(time_diffs),
"max_seconds": np.max(time_diffs),
"mean_minutes": np.mean(time_diffs) / 60,
"target_minutes": 10.0 # 比特幣目標區塊時間
}
def calculate_throughput(self, blocks: List[BlockData], period_hours: int = 24) -> Dict:
"""計算網路吞吐量"""
# 估算每秒交易數 (TPS)
total_tx = sum(b.transaction_count for b in blocks)
# 計算時間範圍
timestamps = [b.timestamp for b in blocks]
time_range = max(timestamps) - min(timestamps)
# 轉換為小時
hours = time_range / 3600
return {
"total_transactions": total_tx,
"period_hours": hours,
"tps_average": total_tx / time_range if time_range > 0 else 0,
"tps_estimated_daily": total_tx / hours if hours > 0 else 0,
"block_rate_per_hour": len(blocks) / hours if hours > 0 else 0
}
def generate_comprehensive_report(self, num_blocks: int = 144) -> Dict:
"""生成綜合報告"""
blocks = self.fetch_recent_blocks(num_blocks)
if not blocks:
return {"error": "無法獲取區塊數據"}
return {
"block_size_stats": self.calculate_block_size_stats(blocks),
"transaction_count_stats": self.calculate_transaction_count_stats(blocks),
"block_time_stats": self.calculate_block_time_stats(blocks),
"throughput": self.calculate_throughput(blocks),
"sample_size": len(blocks),
"height_range": {
"start": blocks[-1].height,
"end": blocks[0].height
}
}
# 使用範例
if __name__ == "__main__":
# 需要先配置 Bitcoin Core RPC
from bitcoin_rpc_client import BitcoinRPCClient
try:
rpc = BitcoinRPCClient(
rpc_user="bitcoin",
rpc_password="your_password"
)
calculator = OnChainMetricsCalculator(rpc)
# 生成報告
report = calculator.generate_comprehensive_report(144)
print("=" * 60)
print("比特幣鏈上指標報告")
print("=" * 60)
print("\n區塊大小統計 (bytes)：")
for key, value in report['block_size_stats'].items():
print(f" {key}: {value:,.0f}")
print("\n每區塊交易數統計：")
for key, value in report['transaction_count_stats'].items():
print(f" {key}: {value:,.0f}")
print("\n區塊時間統計 (秒)：")
for key, value in report['block_time_stats'].items():
print(f" {key}: {value:.2f}")
print("\n網路吞吐量：")
for key, value in report['throughput'].items():
print(f" {key}: {value:.2f}")
except Exception as e:
print(f"錯誤: {e}")進階指標：UTXO 分析
UTXO（未花費交易輸出）模型是比特幣獨特的帳戶系統，分析 UTXO 集合可以提供有價值的市場洞察。
#!/usr/bin/env python3
"""
比特幣 UTXO 分析模組
"""
import requests
import pandas as pd
import numpy as np
from typing import Dict, List, Tuple, Optional
from datetime import datetime
from collections import defaultdict
import time
class UTXOAnalyzer:
"""比特幣 UTXO 分析器"""
def __init__(self, provider="blockstream"):
self.provider = provider
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'BitcoinUTXOAnalyzer/1.0'
})
def get_utxo_set_summary(self) -> Dict:
"""獲取 UTXO 集合摘要"""
if self.provider == "blockstream":
url = "https://blockstream.info/api/utxo/set-summary"
elif self.provider == "blockchain.info":
url = "https://blockchain.info/utxo?active"
else:
raise ValueError(f"Unknown provider: {self.provider}")
response = self.session.get(url, timeout=30)
return response.json()
def get_address_utxo(self, address: str) -> List[Dict]:
"""獲取地址的 UTXO"""
if self.provider == "blockstream":
url = f"https://blockstream.info/api/address/{address}/utxo"
elif self.provider == "blockchain.info":
url = f"https://blockchain.info/unspent?active={address}"
else:
raise ValueError(f"Unknown provider: {self.provider}")
response = self.session.get(url, timeout=30)
return response.json()
def calculate_address_statistics(self, address: str) -> Dict:
"""計算地址統計信息"""
utxos = self.get_address_utxo(address)
if not utxos:
return {
"address": address,
"utxo_count": 0,
"total_satoshi": 0,
"total_btc": 0,
"avg_utxo_size_btc": 0,
"median_utxo_size_btc": 0
}
# 計算總價值
total_satoshi = sum(utxo.get('value', 0) for utxo in utxos)
# 計算每個 UTXO 的價值
values = [utxo.get('value', 0) / 1e8 for utxo in utxos]
return {
"address": address,
"utxo_count": len(utxos),
"total_satoshi": total_satoshi,
"total_btc": total_satoshi / 1e8,
"avg_utxo_size_btc": np.mean(values),
"median_utxo_size_btc": np.median(values),
"min_utxo_size_btc": min(values),
"max_utxo_size_btc": max(values),
"std_utxo_size_btc": np.std(values)
}
def analyze_utxo_distribution(self, addresses: List[str]) -> Dict:
"""分析多個地址的 UTXO 分佈"""
all_utxos = []
address_stats = []
for addr in addresses:
try:
stats = self.calculate_address_statistics(addr)
address_stats.append(stats)
utxos = self.get_address_utxo(addr)
for utxo in utxos:
all_utxos.append({
'address': addr,
'value': utxo.get('value', 0),
'confirmations': utxo.get('confirmations', 0)
})
time.sleep(0.5) # 避免過度請求
except Exception as e:
print(f"處理地址 {addr} 時出錯: {e}")
continue
if not all_utxos:
return {"error": "無法獲取 UTXO 數據"}
# 計算總體統計
values = [utxo['value'] / 1e8 for utxo in all_utxos]
return {
"total_addresses": len(address_stats),
"total_utxos": len(all_utxos),
"total_value_btc": sum(values),
"utxo_value_distribution": {
"mean": np.mean(values),
"median": np.median(values),
"std": np.std(values),
"min": min(values),
"max": max(values),
"p10": np.percentile(values, 10),
"p25": np.percentile(values, 25),
"p75": np.percentile(values, 75),
"p90": np.percentile(values, 90),
"p95": np.percentile(values, 95)
},
"utxo_size_distribution": {
"< 0.001 BTC": len([v for v in values if v < 0.001]),
"0.001-0.01 BTC": len([v for v in values if 0.001 <= v < 0.01]),
"0.01-0.1 BTC": len([v for v in values if 0.01 <= v < 0.1]),
"0.1-1 BTC": len([v for v in values if 0.1 <= v < 1]),
"1-10 BTC": len([v for v in values if 1 <= v < 10]),
">= 10 BTC": len([v for v in values if v >= 10])
},
"confirmation_distribution": {
"unconfirmed": len([u for u in all_utxos if u['confirmations'] == 0]),
"1-6 confirmations": len([u for u in all_utxos if 1 <= u['confirmations'] < 6]),
"6+ confirmations": len([u for u in all_utxos if u['confirmations'] >= 6])
}
}
# 使用範例
if __name__ == "__main__":
analyzer = UTXOAnalyzer("blockstream")
# 測試比特幣創世區塊獎勵地址
# 比特幣第一個區塊的獎勵發送到 1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa
test_address = "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa"
try:
stats = analyzer.calculate_address_statistics(test_address)
print("=" * 60)
print(f"地址 UTXO 分析: {test_address}")
print("=" * 60)
print(f"UTXO 數量: {stats['utxo_count']}")
print(f"總價值: {stats['total_btc']:.8f} BTC")
print(f"平均 UTXO 大小: {stats['avg_utxo_size_btc']:.8f} BTC")
print(f"中位數 UTXO 大小: {stats['median_utxo_size_btc']:.8f} BTC")
# 測試多地址分析
# 比特幣富豪榜前 10 地址
rich_addresses = [
"1BvBMSEYstWetqTFn5Au4m4GFg7xJaNVN2",
"1NDyJtNTjmBW5ZRuxr3Rjmv1fn7cY7MN6n",
"3D2oet4NuZqsqCaL41d51Cdqc4nyP5t1S3",
"bc1q9vza2e8x572nk65f6s7m3dktsq4fpx9lmqhpyy",
"1AC4fMwgY8Kj3L4qY9qZhZbN3R8xN2vT5K",
"1K56i67NFR2nJVCKfWq7oYqYq3XvN9mP6R",
"bc1qar0srrr7xfkvy5l643lydnw9re59gtzzwf5mdq",
"bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4",
"1APM3Z4eJ5gT9qL8rX6vN2hK3fW1yB7tP",
"3Cbq7aT1tY8k4W6nL9hR2vN3pX5zA8fG1"
]
print("\n正在分析多地址 UTXO 分佈...")
distribution = analyzer.analyze_utxo_distribution(rich_addresss[:3])
print("\nUTXO 分佈統計：")
print(f"總地址數: {distribution.get('total_addresses', 0)}")
print(f"總 UTXO 數: {distribution.get('total_utxos', 0)}")
print(f"總價值: {distribution.get('total_value_btc', 0):.2f} BTC")
print("\nUTXO 價值分佈：")
dist = distribution.get('utxo_value_distribution', {})
for key, value in dist.items():
print(f" {key}: {value:.6f}")
except Exception as e:
print(f"錯誤: {e}")比特幣價格與鏈上數據整合分析
比特幣數據儀表板實作
以下程式碼展示如何將比特幣區塊鏈數據與價格數據整合，創建一個簡易的數據儀表板。
#!/usr/bin/env python3
"""
比特幣綜合數據儀表板
整合區塊鏈數據與市場數據
"""
import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List
import time
import json
class BitcoinDashboard:
"""比特幣綜合數據儀表板"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'BitcoinDashboard/1.0'
})
self.cache = {}
self.cache_duration = 60 # 秒
def _get_cached(self, key: str):
"""獲取緩存數據"""
if key in self.cache:
data, timestamp = self.cache[key]
if time.time() - timestamp < self.cache_duration:
return data
return None
def _set_cached(self, key: str, data):
"""設置緩存數據"""
self.cache[key] = (data, time.time())
def get_price_data(self, days: int = 30) -> pd.DataFrame:
"""獲取比特幣價格數據"""
cache_key = f"price_{days}"
cached = self._get_cached(cache_key)
if cached is not None:
return cached
try:
# 使用 CoinGecko API
url = "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart"
params = {
"vs_currency": "usd",
"days": days,
"interval": "daily"
}
response = self.session.get(url, params=params, timeout=30)
data = response.json()
prices = data['market_caps']
df = pd.DataFrame(prices, columns=['timestamp', 'market_cap'])
df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
df['price'] = df['market_cap'] / 19.85e6 # 估算價格
df = df.set_index('date')
df = df.drop('timestamp', axis=1)
self._set_cached(cache_key, df)
return df
except Exception as e:
print(f"獲取價格數據失敗: {e}")
return pd.DataFrame()
def get_blockchain_data(self) -> Dict:
"""獲取區塊鏈統計數據"""
cache_key = "blockchain_stats"
cached = self._get_cached(cache_key)
if cached is not None:
return cached
try:
# 使用 Blockchain.info API
url = "https://blockchain.info/stats?format=json"
response = self.session.get(url, timeout=30)
data = response.json()
stats = {
"block_height": data.get('n_blocks_total', 0),
"total_transactions": data.get('total_transactions_total', 0),
"total_btc_mined": data.get('totalbc', 0) / 1e8,
"market_cap_usd": data.get('market_cap_usd', 0),
"hash_rate_ths": data.get('hash_rate', 0) / 1e12,
"difficulty": data.get('difficulty', 0),
"btc_per_block": data.get('bcd_per_block', 6.25),
"minutes_between_blocks": data.get('minutes_between_blocks', 10)
}
self._set_cached(cache_key, stats)
return stats
except Exception as e:
print(f"獲取區塊鏈數據失敗: {e}")
return {}
def get_fee_estimates(self) -> Dict:
"""獲取費用估計"""
try:
# 使用 mempool.space API
url = "https://mempool.space/api/v1/fees/recommended"
response = self.session.get(url, timeout=30)
return response.json()
except:
# 備用 Blockchain.info
try:
url = "https://blockchain.info/estimatefee?conf=1"
response = self.session.get(url, timeout=30)
data = response.json()
return {
"fastestFee": int(data.get(1, 10) * 1e5),
"halfHourFee": int(data.get(3, 5) * 1e5),
"hourFee": int(data.get(6, 3) * 1e5)
}
except:
return {"fastestFee": 20, "halfHourFee": 10, "hourFee": 5}
def get_mempool_stats(self) -> Dict:
"""獲取內存池統計"""
try:
url = "https://mempool.space/api/mempool"
response = self.session.get(url, timeout=30)
data = response.json()
return {
"tx_count": data.get('count', 0),
"vsize": data.get('vsize', 0),
"total_fee": data.get('total_fee', 0)
}
except:
return {"tx_count": 0, "vsize": 0, "total_fee": 0}
def generate_dashboard_report(self) -> Dict:
"""生成儀表板報告"""
# 獲取各項數據
blockchain_data = self.get_blockchain_data()
price_data = self.get_price_data(30)
fee_estimates = self.get_fee_estimates()
mempool_stats = self.get_mempool_stats()
# 計算價格統計
price_stats = {}
if not price_data.empty:
price_stats = {
"current_price": price_data['price'].iloc[-1],
"price_30d_ago": price_data['price'].iloc[0],
"price_change_30d_pct": ((price_data['price'].iloc[-1] / price_data['price'].iloc[0]) - 1) * 100,
"price_high_30d": price_data['price'].max(),
"price_low_30d": price_data['price'].min(),
"price_volatility_30d": price_data['price'].std() / price_data['price'].mean() * 100
}
return {
"timestamp": datetime.now().isoformat(),
"blockchain": blockchain_data,
"price": price_stats,
"fees": fee_estimates,
"mempool": mempool_stats
}
def print_dashboard(self):
"""打印儀表板"""
report = self.generate_dashboard_report()
print("=" * 70)
print("比特幣綜合數據儀表板")
print("=" * 70)
# 區塊鏈數據
bc = report.get('blockchain', {})
print("\n【區塊鏈數據】")
print(f" 區塊高度: {bc.get('block_height', 'N/A'):,}")
print(f" 總交易數: {bc.get('total_transactions', 'N/A'):,}")
print(f" 總開採比特幣: {bc.get('total_btc_mined', 'N/A'):,.2f} BTC")
print(f" 網路算力: {bc.get('hash_rate_ths', 'N/A'):,.2f} TH/s")
print(f" 挖礦難度: {bc.get('difficulty', 'N/A'):,.0f}")
# 價格數據
ps = report.get('price', {})
print("\n【價格數據 (30天)】")
print(f" 當前價格: ${ps.get('current_price', 'N/A'):,.2f}")
print(f" 30天前價格: ${ps.get('price_30d_ago', 'N/A'):,.2f}")
print(f" 30天變化: {ps.get('price_change_30d_pct', 'N/A'):+.2f}%")
print(f" 30天最高: ${ps.get('price_high_30d', 'N/A'):,.2f}")
print(f" 30天最低: ${ps.get('price_low_30d', 'N/A'):,.2f}")
print(f" 波動率: {ps.get('price_volatility_30d', 'N/A'):.2f}%")
# 費用數據
fees = report.get('fees', {})
print("\n【費用估計 (sat/vB)】")
print(f" 快速 (< 10 分鐘): {fees.get('fastestFee', 'N/A')} sat/vB")
print(f" 中等 (< 30 分鐘): {fees.get('halfHourFee', 'N/A')} sat/vB")
print(f" 慢速 (< 1 小時): {fees.get('hourFee', 'N/A')} sat/vB")
# 內存池數據
mp = report.get('mempool', {})
print("\n【內存池狀態】")
print(f" 交易數量: {mp.get('tx_count', 'N/A'):,}")
print(f" 總大小: {mp.get('vsize', 'N/A'):,} vbytes")
print("\n" + "=" * 70)
print(f"最後更新: {report.get('timestamp', 'N/A')}")
print("=" * 70)
# 使用範例
if __name__ == "__main__":
dashboard = BitcoinDashboard()
dashboard.print_dashboard()總結
本文提供了比特幣區塊鏈數據分析的完整實作教學，涵蓋了以下關鍵主題：
第一，比特幣區塊結構的詳細解析，包括區塊頭、交易結構和關鍵字段的說明。
第二，多種比特幣數據獲取方式的實作，包括 Bitcoin Core RPC、公共 API 服務（Blockchain.info、Blockstream、Mempool.space）等。
第三，鏈上指標的計算方法，包括區塊大小統計、交易數量分析、網路吞吐量和 UTXO 分析等。
第四，比特幣數據儀表板的整合實現，展示如何將多個數據源整合成有意義的分析報告。
這些工具和技術可以應用於多種場景，包括市場研究、風險管理、投資決策支援和學術研究等。讀者可以根據自己的需求進一步擴展和定制這些程式碼。
參考資源
API 文档
- Bitcoin Core RPC API: https://developer.bitcoin.org/reference/rpc/
- Blockchain.info API: https://www.blockchain.com/explorer/api
- Blockstream API: https://blockstream.info/api/
- Mempool.space API: https://mempool.space/api/
- CoinGecko API: www.coingecko.com/en/api
開源專案
- Bitcoin Core: https://github.com/bitcoin/bitcoin
- btc-rpc-explorer: https://github.com/janoside/btc-rpc-explorer
學習資源
- Mastering Bitcoin (Andreas Antonopoulos)
- Bitcoin Developer Documentation
更新日期：2026-03-04
版本：1.0
相關文章
- Bitcoin Core 節點運作 — 運行完整節點，理解比特幣網路的運作機制。
- 比特幣區塊鏈數據分析實作指南 — 完整的比特幣區塊鏈數據分析方法論教學，涵蓋 Python 數據獲取、價格分析、減半週期研究、技術指標計算與鏈上指標實作，提供可直接執行的程式碼範例。
- 比特幣鏈上指標深度分析：從數據獲取到投資決策的完整實務指南 — 深入介紹比特幣鏈上指標的理論基礎、計算方法、數據來源，提供可直接執行的 Python 程式碼範例，涵蓋 NVT、MVRV、庫存流量模型等關鍵指標的分析與應用。
- 比特幣治理深度分析 — 比特幣網路治理機制分析
- 比特幣即時數據 API 整合教學 — 比特幣區塊鏈 API 整合的完整指南，從節點數量、網路算力到記憶池狀態的即時監控實作，包含代碼範例與費用預測模型。
延伸閱讀與來源
這篇文章對您有幫助嗎？
請告訴我們如何改進：
0 人覺得有帮助
評論
發表評論
注意：由於這是靜態網站，您的評論將儲存在本地瀏覽器中，不會公開顯示。
目前尚無評論，成為第一個發表評論的人吧！