比特幣區塊鏈數據分析實作指南
完整的比特幣區塊鏈數據分析方法論教學,涵蓋 Python 數據獲取、價格分析、減半週期研究、技術指標計算與鏈上指標實作,提供可直接執行的程式碼範例。
比特幣區塊鏈數據分析實作指南
比特幣區塊鏈蘊含豐富的數據,對這些數據進行分析可以獲得有價值的市場洞察。本文提供完整的比特幣區塊鏈數據分析方法論,包括使用 Python 進行數據獲取、分析和可視化的實作範例。我們將涵蓋鏈上數據分析、交易模式識別和市場情緒指標計算等主題。
環境設置與數據獲取
必要的 Python 環境
首先,我們需要設置開發環境並安裝必要的 Python 庫:
# 創建虛擬環境
python -m venv bitcoin_analysis
source bitcoin_analysis/bin/activate # Linux/Mac
# 或
bitcoin_analysis\Scripts\activate # Windows
# 安裝必要的庫
pip install requests pandas numpy matplotlib seaborn plotly
pip install scipy statsmodels
pip install python-bitcoinlib # 比特幣數據處理
pip install web3.py # 以太坊兼容接口
比特幣 API 接口
有多種方式可以獲取比特幣區塊鏈數據:
import requests
import pandas as pd
from datetime import datetime, timedelta
# 使用 mempool.space API(開源、社區驅動的比特區塊瀏覽器)
class MempoolSpaceAPI:
BASE_URL = "https://mempool.space/api"
@staticmethod
def get_mempool_stats():
"""獲取記憶池統計數據"""
response = requests.get(f"{MempoolSpaceAPI.BASE_URL}/mempool")
return response.json()
@staticmethod
def get_fee_estimates():
"""獲取費用估算數據"""
response = requests.get(f"{MempoolSpaceAPI.BASE_URL}/fees/recommended")
return response.json()
@staticmethod
def get_block_height():
"""獲取最新區塊高度"""
response = requests.get(f"{MempoolSpaceAPI.BASE_URL}/blocks/tip/height")
return int(response.text)
@staticmethod
def get_block_hash(height):
"""獲取特定高度的區塊哈希"""
response = requests.get(f"{MempoolSpaceAPI.BASE_URL}/block-height/{height}")
return response.text
@staticmethod
def get_block_data(block_hash):
"""獲取區塊詳細數據"""
response = requests.get(f"{MempoolSpaceAPI.BASE_URL}/block/{block_hash}")
return response.json()
# 使用示例
api = MempoolSpaceAPI()
fee_estimates = api.get_fee_estimates()
print(f"快速確認費用: {fee_estimates['fastestFee']} sat/vB")
print(f"中等確認費用: {fee_estimates['hourFee']} sat/vB")
print(f"低優先級費用: {fee_estimates['minimumFee']} sat/vB")
比特幣區塊數據獲取
import requests
import time
from typing import List, Dict
class BitcoinBlockchainData:
"""比特幣區塊鏈數據獲取類"""
def __init__(self):
self.base_url = "https://blockstream.info/api"
def get_block_by_height(self, height: int) -> Dict:
"""通過區塊高度獲取區塊數據"""
url = f"{self.base_url}/block/{height}"
response = requests.get(url)
return response.json()
def get_block_transactions(self, block_hash: str, start: int = 0) -> List[Dict]:
"""獲取區塊中的交易"""
url = f"{self.base_url}/block/{block_hash}/txs/{start}"
response = requests.get(url)
return response.json()
def get_transaction(self, txid: str) -> Dict:
"""獲取特定交易的詳細信息"""
url = f"{self.base_url}/tx/{txid}"
response = requests.get(url)
return response.json()
def get_address_info(self, address: str) -> Dict:
"""獲取地址信息"""
url = f"{self.base_url}/address/{address}"
response = requests.get(url)
return response.json()
def get_address_transactions(self, address: str) -> List[Dict]:
"""獲取地址的所有交易"""
url = f"{self.base_url}/address/{address}/txs"
response = requests.get(url)
return response.json()
# 使用示例
btc = BitcoinBlockchainData()
latest_height = 870000 # 假設的最新區塊高度
# 獲取區塊數據
block_data = btc.get_block_by_height(latest_height)
print(f"區塊高度: {block_data['height']}")
print(f"區塊哈希: {block_data['id']}")
print(f"交易數量: {block_data['tx_count']}")
print(f"區塊大小: {block_data['size'] / 1024:.2f} KB")
print(f"區塊獎勵: {block_data['reward'] / 100000000} BTC")
基礎數據分析
比特幣價格與時間關係分析
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import requests
def fetch_btc_price_history(days: int = 365) -> pd.DataFrame:
"""從 CoinGecko API 獲取比特幣歷史價格數據"""
url = "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart"
params = {
'vs_currency': 'usd',
'days': days,
'interval': 'daily'
}
response = requests.get(url, params=params)
data = response.json()
df = pd.DataFrame(data['prices'], columns=['timestamp', 'price'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
# 獲取一年的比特幣價格數據
btc_price = fetch_btc_price_history(365)
# 計算基本統計指標
print("比特幣價格基本統計(過去一年):")
print(f"平均價格: ${btc_price['price'].mean():,.2f}")
print(f"最高價格: ${btc_price['price'].max():,.2f}")
print(f"最低價格: ${btc_price['price'].min():,.2f}")
print(f"當前價格: ${btc_price['price'].iloc[-1]:,.2f}")
print(f"年化收益率: {((btc_price['price'].iloc[-1] / btc_price['price'].iloc[0]) - 1) * 100:.2f}%")
# 計算收益率
btc_price['returns'] = btc_price['price'].pct_change()
print(f"日均收益率: {btc_price['returns'].mean() * 100:.4f}%")
print(f"收益率標準差: {btc_price['returns'].std() * 100:.4f}%")
print(f"年化波動率: {btc_price['returns'].std() * np.sqrt(365) * 100:.2f}%")
減半週期數據分析
import pandas as pd
import numpy as np
# 定義比特幣減半日期與區塊高度
halving_dates = pd.DataFrame({
'halving': [1, 2, 3, 4],
'date': pd.to_datetime([
'2012-11-28',
'2016-07-09',
'2020-05-11',
'2024-04-20'
]),
'block_height': [210000, 420000, 630000, 840000],
'reward_before': [50, 25, 12.5, 6.25],
'reward_after': [25, 12.5, 6.25, 3.125]
})
# 減半週期分析
def analyze_halving_cycle(df: pd.DataFrame, halving_df: pd.DataFrame) -> pd.DataFrame:
"""分析各次減半週期的價格表現"""
results = []
for i, halving in halving_df.iterrows():
# 減半前 180 天
start_date = halving['date'] - pd.Timedelta(days=180)
# 減半後 180 天
end_date = halving['date'] + pd.Timedelta(days=180)
# 篩選相關時期的數據
period_data = df[(df.index >= start_date) & (df.index <= end_date)]
if len(period_data) > 0:
pre_halving = df[(df.index >= start_date) & (df.index < halving['date'])]
post_halving = df[(df.index >= halving['date']) & (df.index <= end_date)]
results.append({
'halving': halving['halving'],
'date': halving['date'],
'pre_180d_return': ((pre_halving['price'].iloc[-1] / pre_halving['price'].iloc[0]) - 1) * 100
if len(pre_halving) > 1 else 0,
'post_180d_return': ((post_halving['price'].iloc[-1] / post_halving['price'].iloc[0]) - 1) * 100
if len(post_halving) > 1 else 0,
'pre_avg_volatility': pre_halving['returns'].std() * np.sqrt(365) * 100
if len(pre_halving) > 1 else 0,
'post_avg_volatility': post_halving['returns'].std() * np.sqrt(365) * 100
if len(post_halving) > 1 else 0,
})
return pd.DataFrame(results)
# 執行減半週期分析
halving_analysis = analyze_halving_cycle(btc_price, halving_dates)
print("比特幣減半週期分析結果:")
print(halving_analysis.to_string(index=False))
鏈上數據分析
import requests
import pandas as pd
from datetime import datetime, timedelta
class OnChainData:
"""鏈上數據分析類"""
def __init__(self):
# 使用不同數據源
self.blockstream_url = "https://blockstream.info/api"
self.glassnode_url = "https://api.glassnode.com/v1"
def get_market_data(self) -> pd.DataFrame:
"""獲取市場數據"""
# 獲取庫存流量比等基礎數據
# 2100萬總供應量
total_supply = 21_000_000
# 計算每日新增供應(基於區塊獎勵)
# 假設每日產出約 144 個區塊(每 10 分鐘一個)
blocks_per_day = 144
# 根據當前區塊高度估算區塊獎勵
current_block = 870000 # 假設當前高度
halving_interval = 210000
# 計算當前減半週期
halving_count = current_block // halving_interval
current_reward = 50 / (2 ** halving_count)
daily_new_supply = blocks_per_day * current_reward
# Stock-to-Flow 比率
annual_supply = daily_new_supply * 365
s2f = total_supply / annual_supply
print(f"當前區塊獎勵: {current_reward} BTC")
print(f"每日新供應: {daily_new_supply:.2f} BTC")
print(f"年通脹率: {(annual_supply / total_supply) * 100:.2f}%")
print(f"庫存流量比 (S2F): {s2f:.2f}")
return {
'current_reward': current_reward,
'daily_supply': daily_new_supply,
'annual_inflation': (annual_supply / total_supply) * 100,
'stock_to_flow': s2f
}
def get_hashrate_data(self) -> dict:
"""獲取算力數據"""
# 獲取難度調整數據
response = requests.get(f"{self.blockstream_url}/stats")
stats = response.json()
return {
'difficulty': stats.get('difficulty'),
'hashrate': stats.get('estimated_hashrate'),
'block_height': stats.get('blocks')
}
# 使用示例
onchain = OnChainData()
market_data = onchain.get_market_data()
hashrate_data = onchain.get_hashrate_data()
print(f"\n算力數據:")
print(f"當前難度: {hashrate_data['difficulty']:,.2f}")
print(f"估算算力: {hashrate_data['hashrate']:,.2f} H/s")
進階分析方法
波動率分析
import pandas as pd
import numpy as np
from scipy import stats
def calculate_volatility_metrics(price_data: pd.DataFrame, window: int = 30) -> pd.DataFrame:
"""計算波動率指標"""
df = price_data.copy()
# 日收益率
df['returns'] = df['price'].pct_change()
# 歷史波動率(滾動窗口)
df['historical_volatility'] = df['returns'].rolling(window=window).std() * np.sqrt(365)
# 指數加權移動平均波動率
df['ewm_volatility'] = df['returns'].ewm(span=window).std() * np.sqrt(365)
# 波動率比率(相對於歷史平均)
avg_volatility = df['historical_volatility'].mean()
df['volatility_ratio'] = df['historical_volatility'] / avg_volatility
return df
# 計算波動率指標
btc_volatility = calculate_volatility_metrics(btc_price)
print("比特幣波動率分析結果:")
print(f"30日滾動年化波動率(最新): {btc_volatility['historical_volatility'].iloc[-1] * 100:.2f}%")
print(f"平均年化波動率: {btc_volatility['historical_volatility'].mean() * 100:.2f}%")
print(f"波動率比率(最新): {btc_volatility['volatility_ratio'].iloc[-1]:.2f}x")
# 波動率分位數分析
vol_percentiles = btc_volatility['historical_volatility'].quantile([0.1, 0.25, 0.5, 0.75, 0.9])
print("\n波動率分位數(年化):")
print(f"10% 分位數: {vol_percentiles[0.1] * 100:.2f}%")
print(f"50% 分位數: {vol_percentiles[0.5] * 100:.2f}%")
print(f"90% 分位數: {vol_percentiles[0.9] * 100:.2f}%")
技術指標計算
import pandas as pd
import numpy as np
def calculate_technical_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""計算比特幣技術分析指標"""
# 簡單移動平均 (SMA)
df['SMA_20'] = df['price'].rolling(window=20).mean()
df['SMA_50'] = df['price'].rolling(window=50).mean()
df['SMA_200'] = df['price'].rolling(window=200).mean()
# 指數移動平均 (EMA)
df['EMA_12'] = df['price'].ewm(span=12, adjust=False).mean()
df['EMA_26'] = df['price'].ewm(span=26, adjust=False).mean()
# MACD
df['MACD'] = df['EMA_12'] - df['EMA_26']
df['MACD_signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
df['MACD_histogram'] = df['MACD'] - df['MACD_signal']
# RSI (Relative Strength Index)
delta = df['price'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
# 布林帶 (Bollinger Bands)
df['BB_middle'] = df['price'].rolling(window=20).mean()
bb_std = df['price'].rolling(window=20).std()
df['BB_upper'] = df['BB_middle'] + (bb_std * 2)
df['BB_lower'] = df['BB_middle'] - (bb_std * 2)
df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle']
# ATR (Average True Range)
high_low = df['price'].diff().abs()
high_close = np.abs(df['price'].diff())
low_close = np.abs(df['price'].diff())
ranges = pd.concat([high_low, high_close, low_close], axis=1)
true_range = ranges.max(axis=1)
df['ATR'] = true_range.rolling(14).mean()
df['ATR_percent'] = df['ATR'] / df['price'] * 100
return df
# 計算技術指標
btc_tech = calculate_technical_indicators(btc_price)
# 輸出最新技術指標
latest = btc_tech.iloc[-1]
print("比特幣技術指標(最新):")
print(f"價格: ${latest['price']:,.2f}")
print(f"SMA 20: ${latest['SMA_20']:,.2f}")
print(f"SMA 50: ${latest['SMA_50']:,.2f}")
print(f"SMA 200: ${latest['SMA_200']:,.2f}")
print(f"RSI (14): {latest['RSI']:.2f}")
print(f"MACD: {latest['MACD']:.2f}")
print(f"MACD Signal: {latest['MACD_signal']:.2f}")
print(f"布林帶寬度: {latest['BB_width']*100:.2f}%")
print(f"ATR %: {latest['ATR_percent']:.2f}%")
# 技術信號識別
print("\n技術信號識別:")
if latest['price'] > latest['SMA_200']:
print("✓ 價格高於 SMA 200 - 長期看漲")
else:
print("✗ 價格低於 SMA 200 - 長期看跌")
if latest['RSI'] > 70:
print("⚠ RSI 超買 - 可能回調")
elif latest['RSI'] < 30:
print("⚠ RSI 超賣 - 可能反彈")
else:
print("✓ RSI 中性")
if latest['MACD'] > latest['MACD_signal']:
print("✓ MACD 金叉 - 短期看漲")
else:
print("✗ MACD 死叉 - 短期看跌")
鏈上活躍度指標
import requests
import pandas as pd
import numpy as np
class OnChainMetrics:
"""鏈上活躍度指標計算"""
@staticmethod
def calculate_nvt_ratio(prices: pd.Series, volume: pd.Series, window: int = 90) -> pd.Series:
"""計算 NVT 比率 (Network Value to Transactions)
NVT 被稱為比特幣的「市盈率」,用於評估比特幣的估值水平
"""
nvt = prices / volume.rolling(window=window).mean()
return nvt
@staticmethod
def calculate_mvrv_ratio(prices: pd.Series, realized_price: pd.Series) -> pd.Series:
"""計算 MVRV 比率 (Market Value to Realized Value)
MVRV > 3.5 表示比特幣被嚴重高估
MVRV < 1 表示比特幣被嚴重低估
"""
return prices / realized_price
@staticmethod
def calculate_hodl_wave(utxo_age_distribution: pd.DataFrame) -> dict:
"""計算 HODL 波
長期持有者的持倉比例變化
"""
# 假設有 UTXO 年齡分布數據
long_term_holders = utxo_age_distribution[utxo_age_distribution['age_days'] > 15552000].sum()
total_utxo = utxo_age_distribution['value'].sum()
hodl_ratio = long_term_holders / total_utxo
return {'hodl_ratio': hodl_ratio, 'long_term_supply': long_term_holders}
# 示範 NVT 比率計算(需要真實交易量數據)
def estimate_transaction_volume(price_data: pd.DataFrame) -> pd.DataFrame:
"""估算鏈上交易量(基於價格和區塊數據)"""
# 假設每個區塊平均交易數
avg_tx_per_block = 2500
# 假設平均交易價值
avg_tx_value = 0.5 # BTC
# 每日區塊數
blocks_per_day = 144
# 估算每日交易量
daily_volume = price_data.copy()
daily_volume['estimated_volume'] = (
avg_tx_per_block *
avg_tx_value *
price_data['price']
)
return daily_volume
# 計算估算的交易量
btc_volume = estimate_transaction_volume(btc_price)
# 計算 NVT 比率(使用估算值)
onchain = OnChainMetrics()
btc_volume['NVT'] = onchain.calculate_nvt_ratio(
btc_volume['price'],
btc_volume['estimated_volume']
)
print("比特幣 NVT 比率(最新):")
print(f"NVT 比率: {btc_volume['NVT'].iloc[-1]:.2f}")
print(f"NVT 比率(90日平均): {btc_volume['NVT'].rolling(90).mean().iloc[-1]:.2f}")
print("\nNVT 比率解釋:")
print("- 低 NVT (<15): 比特幣可能被低估")
print("- 中性 NVT (15-30): 估值合理")
print("- 高 NVT (>30): 比特幣可能被高估")
數據可視化
價格與波動率可視化
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib import style
# 設置繪圖風格
style.use('seaborn-v0_8-whitegrid')
plt.rcParams['figure.figsize'] = (14, 10)
plt.rcParams['font.size'] = 12
# 創建子圖
fig, axes = plt.subplots(3, 1, figsize=(14, 12), sharex=True)
# 1. 比特幣價格走勢圖
ax1 = axes[0]
ax1.plot(btc_price.index, btc_price['price'], 'b-', linewidth=1.5, label='比特幣價格')
ax1.fill_between(btc_price.index, 0, btc_price['price'], alpha=0.3)
ax1.set_ylabel('價格 (USD)', fontsize=12)
ax1.set_title('比特幣價格走勢與技術指標分析', fontsize=14, fontweight='bold')
ax1.legend(loc='upper left')
ax1.set_yscale('log') # 對數刻度
# 添加 SMA 線
ax1.plot(btc_tech.index, btc_tech['SMA_50'], 'orange', linewidth=1, label='SMA 50', alpha=0.8)
ax1.plot(btc_tech.index, btc_tech['SMA_200'], 'red', linewidth=1, label='SMA 200', alpha=0.8)
# 2. RSI 指標
ax2 = axes[1]
ax2.plot(btc_tech.index, btc_tech['RSI'], 'purple', linewidth=1)
ax2.axhline(y=70, color='r', linestyle='--', alpha=0.5, label='超買線')
ax2.axhline(y=30, color='g', linestyle='--', alpha=0.5, label='超賣線')
ax2.axhline(y=50, color='gray', linestyle='-', alpha=0.3)
ax2.fill_between(btc_tech.index, 30, 70, alpha=0.1, color='gray')
ax2.set_ylabel('RSI', fontsize=12)
ax2.set_ylim(0, 100)
ax2.legend(loc='upper left')
ax2.set_title('相對強弱指數 (RSI)', fontsize=12)
# 3. 波動率
ax3 = axes[2]
ax3.plot(btc_volatility.index, btc_volatility['historical_volatility'] * 100, 'r-', linewidth=1.5, label='年化波動率')
ax3.fill_between(btc_volatility.index, 0, btc_volatility['historical_volatility'] * 100, alpha=0.3, color='red')
ax3.axhline(y=btc_volatility['historical_volatility'].mean() * 100, color='blue', linestyle='--', label='平均波動率')
ax3.set_ylabel('波動率 (%)', fontsize=12)
ax3.set_xlabel('日期', fontsize=12)
ax3.legend(loc='upper right')
ax3.set_title('比特幣波動率分析', fontsize=12)
# 格式化 x 軸日期
for ax in axes:
ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))
ax.xaxis.set_major_locator(mdates.MonthLocator(interval=3))
plt.setp(ax.xaxis.get_majorticklabels(), rotation=45)
plt.tight_layout()
plt.savefig('bitcoin_analysis.png', dpi=150, bbox_inches='tight')
plt.show()
print("\n圖表已保存至 bitcoin_analysis.png")
結論與實踐建議
數據分析的關鍵要點
比特幣區塊鏈數據分析需要結合多種數據源和分析方法:
- 數據來源選擇:不同的 API 提供不同的數據範圍和精度,需要根據分析目標選擇合適的數據源。
- 指標理解:每個指標都有其局限性,應該綜合多個指標進行判斷。
- 歷史規律:比特幣市場相對年輕,歷史數據的統計顯著性有限。
- 宏觀因素:比特幣價格受多種因素影響,技術分析只是其中一部分。
進一步學習方向
- 深入學習區塊鏈數據結構
- 掌握機器學習在金融時間序列中的應用
- 學習更多區塊鏈數據 API(Glassnode, Chainalysis 等)
- 研究衍生品市場數據分析
- 實踐比特幣節點運營和區塊數據直接分析
風險提示
數據分析和技術指標僅供參考,不構成投資建議。比特幣投資存在高風險,投資者應根據自身風險承受能力做出決策,並建議諮詢專業財務顧問。
參考資源
- Bitcoin Wiki - Developer Documentation
- Mempool.space API Documentation
- Blockstream.info API Documentation
- CoinGecko API Documentation
- 《比特幣的價格由什麼決定?》- 區塊鏈數據分析入門
- 《Mastering Bitcoin》- Andreas M. Antonopoulos
相關文章
- Bitcoin Core RPC 快速上手 — 透過 JSON-RPC 介面與比特幣節點互動,常用指令教學。
- 比特幣密碼學基礎 — 深入理解比特幣核心密碼學技術:SHA-256、RIPEMD-160、secp256k1 橢圓曲線、ECDSA 與 Schnorr 簽章。
- Nakamoto 共識機制 — 深入分析比特幣的革命性共識機制:工作量證明、最長鏈原則、激勵相容性與安全性分析。
- Taproot 全面解析 — 比特幣最新的腳本升級:MAST、BIP-340/341/342。
- OP_CHECKTEMPLATEVERIFY 深度技術分析 — 深入分析 BIP 119 提出的 CTV 技術原理、應用場景、優勢與風險,以及當前發展狀態。
延伸閱讀與來源
這篇文章對您有幫助嗎?
請告訴我們如何改進:
0 人覺得有帮助
評論
發表評論
注意:由於這是靜態網站,您的評論將儲存在本地瀏覽器中,不會公開顯示。
目前尚無評論,成為第一個發表評論的人吧!