小明的數位黃金時代

昨天我們了解了量化交易的基本概念，今天要深入探討虛擬貨幣的量化交易。還記得村裡第一次有人用手機支付買菜時，大家都覺得很神奇嗎？虛擬貨幣就像是金融世界的智慧型手機，帶來了全新的交易可能性！

虛擬貨幣市場的特殊性

與傳統金融市場的差異

虛擬貨幣市場的優勢

1. 24/7 交易

   • 不受傳統市場開收盤限制
   • 全球化市場，永不停歇
   • 適合自動化交易系統

2. 高波動性

   • 日內波動常超過 10%
   • 提供更多套利機會
   • 風險與機會並存

3. 程式化友好

   • API 接口完善
   • 交易成本相對較低
   • 技術創新活躍

4. 多元化交易對

   • 數千種加密貨幣
   • 各種交易對組合
   • 現貨、期貨、選擇權齊全

虛擬貨幣量化交易策略類型

1. 趨勢跟隨策略

import pandas as pd
import numpy as np

class TrendFollowingStrategy:
    """趨勢跟隨策略 - 適合虛擬貨幣的高波動特性"""
    
    def __init__(self, fast_period=20, slow_period=50):
        self.fast_period = fast_period
        self.slow_period = slow_period
    
    def calculate_signals(self, price_data):
        """計算交易信號"""
        
        # 計算移動平均線
        price_data['fast_ma'] = price_data['close'].rolling(self.fast_period).mean()
        price_data['slow_ma'] = price_data['close'].rolling(self.slow_period).mean()
        
        # 生成信號
        price_data['signal'] = 0
        price_data.loc[price_data['fast_ma'] > price_data['slow_ma'], 'signal'] = 1  # 買入
        price_data.loc[price_data['fast_ma'] < price_data['slow_ma'], 'signal'] = -1 # 賣出
        
        return price_data
    
    def backtest(self, price_data):
        """回測策略表現"""
        
        data = self.calculate_signals(price_data)
        
        # 計算策略收益
        data['returns'] = data['close'].pct_change()
        data['strategy_returns'] = data['signal'].shift(1) * data['returns']
        
        # 計算累積收益
        data['cumulative_returns'] = (1 + data['strategy_returns']).cumprod()
        
        return data

2. 均值回歸策略

虛擬貨幣市場經常出現劇烈波動後的回調：

class MeanReversionStrategy:
    """均值回歸策略 - 利用虛擬貨幣的過度反應特性"""
    
    def __init__(self, lookback_period=20, std_threshold=2):
        self.lookback_period = lookback_period
        self.std_threshold = std_threshold
    
    def calculate_signals(self, price_data):
        """識別過度偏離均值的機會"""
        
        # 計算移動平均和標準差
        price_data['mean'] = price_data['close'].rolling(self.lookback_period).mean()
        price_data['std'] = price_data['close'].rolling(self.lookback_period).std()
        
        # 計算 Z-score
        price_data['z_score'] = (price_data['close'] - price_data['mean']) / price_data['std']
        
        # 生成信號
        price_data['signal'] = 0
        price_data.loc[price_data['z_score'] > self.std_threshold, 'signal'] = -1  # 超買，做空
        price_data.loc[price_data['z_score'] < -self.std_threshold, 'signal'] = 1  # 超賣，做多
        
        return price_data

3. 套利策略

虛擬貨幣市場的效率性較低，存在多種套利機會：

實際案例：期現套利策略

策略原理

期現套利利用現貨價格和期貨價格之間的價差：

class SpotFutureArbitrage:
    """期現套利策略"""
    
    def __init__(self, entry_threshold=0.01, exit_threshold=0.005):
        self.entry_threshold = entry_threshold  # 進場閾值 1%
        self.exit_threshold = exit_threshold    # 出場閾值 0.5%
    
    def calculate_basis(self, spot_price, future_price):
        """計算基差（期貨溢價率）"""
        return (future_price - spot_price) / spot_price
    
    def generate_signals(self, spot_data, future_data):
        """生成套利信號"""
        
        # 確保數據對齊
        combined_data = pd.DataFrame({
            'spot_price': spot_data['close'],
            'future_price': future_data['close'],
            'timestamp': spot_data['timestamp']
        })
        
        # 計算基差
        combined_data['basis'] = self.calculate_basis(
            combined_data['spot_price'], 
            combined_data['future_price']
        )
        
        # 生成信號
        combined_data['signal'] = 0
        
        # 正向套利：期貨溢價過高
        combined_data.loc[
            combined_data['basis'] > self.entry_threshold, 
            'signal'
        ] = 1  # 買現貨，賣期貨
        
        # 反向套利：期貨溢價過低（貼水）
        combined_data.loc[
            combined_data['basis'] < -self.entry_threshold, 
            'signal'
        ] = -1  # 賣現貨，買期貨
        
        # 平倉信號
        combined_data.loc[
            abs(combined_data['basis']) < self.exit_threshold, 
            'signal'
        ] = 0
        
        return combined_data

# 使用示例
arbitrage = SpotFutureArbitrage()

# 模擬數據
spot_data = pd.DataFrame({
    'close': [50000, 50100, 50200, 50150, 50300],
    'timestamp': pd.date_range('2024-01-01', periods=5, freq='1H')
})

future_data = pd.DataFrame({
    'close': [50600, 50700, 50800, 50750, 50900],  # 期貨溢價約 1.2%
    'timestamp': pd.date_range('2024-01-01', periods=5, freq='1H')
})

signals = arbitrage.generate_signals(spot_data, future_data)
print(signals)

虛擬貨幣量化交易的技術架構

數據收集系統

import websocket
import json
import threading
from queue import Queue

class CryptoDataCollector:
    """虛擬貨幣數據收集器"""
    
    def __init__(self, symbols=['BTCUSDT', 'ETHUSDT']):
        self.symbols = symbols
        self.data_queue = Queue()
        self.ws = None
    
    def on_message(self, ws, message):
        """處理 WebSocket 消息"""
        try:
            data = json.loads(message)
            self.data_queue.put(data)
        except json.JSONDecodeError:
            print(f"Invalid JSON: {message}")
    
    def on_error(self, ws, error):
        print(f"WebSocket error: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        print("WebSocket connection closed")
    
    def on_open(self, ws):
        """訂閱市場數據"""
        subscribe_msg = {
            "method": "SUBSCRIBE",
            "params": [f"{symbol.lower()}@ticker" for symbol in self.symbols],
            "id": 1
        }
        ws.send(json.dumps(subscribe_msg))
    
    def start_collecting(self):
        """開始收集數據"""
        websocket.enableTrace(True)
        self.ws = websocket.WebSocketApp(
            "wss://stream.binance.com:9443/ws/",
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        
        # 在背景執行
        thread = threading.Thread(target=self.ws.run_forever)
        thread.daemon = True
        thread.start()
    
    def get_latest_data(self):
        """獲取最新數據"""
        if not self.data_queue.empty():
            return self.data_queue.get()
        return None

風險管理系統

class CryptoRiskManager:
    """虛擬貨幣風險管理系統"""
    
    def __init__(self, max_position_size=0.1, max_drawdown=0.2):
        self.max_position_size = max_position_size    # 最大倉位 10%
        self.max_drawdown = max_drawdown              # 最大回撤 20%
        self.current_positions = {}
        self.portfolio_value = 100000  # 初始資金 10 萬
        self.peak_value = 100000
    
    def calculate_position_size(self, symbol, price, confidence):
        """計算倉位大小"""
        
        # 基於信心度調整倉位
        base_position = self.max_position_size * confidence
        
        # 考慮虛擬貨幣的高波動性，降低倉位
        volatility_adjustment = 0.5  # 50% 調整係數
        adjusted_position = base_position * volatility_adjustment
        
        # 計算具體數量
        max_value = self.portfolio_value * adjusted_position
        quantity = max_value / price
        
        return quantity
    
    def check_risk_limits(self, symbol, new_position):
        """檢查風險限制"""
        
        # 檢查單一資產集中度
        total_value = sum(pos['value'] for pos in self.current_positions.values())
        if new_position['value'] / total_value > self.max_position_size:
            return False, "超過單一資產最大倉位限制"
        
        # 檢查總體回撤
        current_drawdown = (self.peak_value - self.portfolio_value) / self.peak_value
        if current_drawdown > self.max_drawdown:
            return False, f"當前回撤 {current_drawdown:.2%} 超過限制 {self.max_drawdown:.2%}"
        
        return True, "風險檢查通過"
    
    def update_portfolio(self, symbol, price, quantity):
        """更新投資組合"""
        
        value = price * quantity
        self.current_positions[symbol] = {
            'quantity': quantity,
            'price': price,
            'value': value
        }
        
        # 更新投資組合總值
        self.portfolio_value = sum(pos['value'] for pos in self.current_positions.values())
        
        # 更新峰值
        if self.portfolio_value > self.peak_value:
            self.peak_value = self.portfolio_value

虛擬貨幣量化交易的挑戰

1. 極高波動性

2. 市場操縱風險

虛擬貨幣市場相對年輕，存在操縱風險：

• 鯨魚交易者：大戶可能影響價格
• 消息面操縱：假消息容易影響市場
• 薄弱的流動性：小幣種容易被操縱

3. 技術風險

• 交易所風險：交易所可能停機或倒閉
• 網路風險：網路延遲影響交易執行
• API 限制：請求頻率限制

最佳實踐建議

1. 多交易所策略

class MultiExchangeStrategy:
    """多交易所策略 - 降低單點風險"""
    
    def __init__(self, exchanges=['binance', 'bybit', 'okx']):
        self.exchanges = exchanges
        self.connections = {}
    
    def connect_all_exchanges(self):
        """連接所有交易所"""
        for exchange in self.exchanges:
            self.connections[exchange] = self.create_connection(exchange)
    
    def get_best_price(self, symbol, side):
        """獲取最佳價格"""
        prices = {}
        for exchange in self.exchanges:
            try:
                price = self.connections[exchange].get_price(symbol, side)
                prices[exchange] = price
            except Exception as e:
                print(f"{exchange} 價格獲取失敗: {e}")
        
        if side == 'buy':
            return min(prices.items(), key=lambda x: x[1])  # 最低買價
        else:
            return max(prices.items(), key=lambda x: x[1])  # 最高賣價

2. 階段式倉位管理

def staged_position_management(price_change, base_position):
    """階段式倉位管理 - 適應高波動性"""
    
    if abs(price_change) < 0.05:      # 5% 以內正常倉位
        return base_position
    elif abs(price_change) < 0.10:    # 5-10% 減少一半倉位
        return base_position * 0.5
    elif abs(price_change) < 0.20:    # 10-20% 大幅減少倉位
        return base_position * 0.2
    else:                             # 超過 20% 停止交易
        return 0

小結

今天我們深入探討了虛擬貨幣的量化交易特性，就像認識了一個全新的農業領域。虛擬貨幣市場的特點：

優勢：

• 24/7 交易機會
• 高波動性提供更多機會
• 程式化交易友好
• 進入門檻相對較低

挑戰：

• 極高的風險
• 市場不成熟
• 技術和監管風險

成功關鍵：

• 嚴格的風險管理
• 多元化策略
• 持續學習和適應

明天我們將比較加密貨幣與傳統市場的差異，了解如何在這個新興市場中找到機會！

下一篇：Day 18 - 加密貨幣 vs 傳統市場