今天要學習 Paper Trade（模擬交易），這就像爸爸在決定種新作物前，先在小塊試驗田做實驗一樣。在真正投入資金之前，我們需要在安全的環境中驗證策略，累積經驗，就像先用假錢練習，等熟練了再用真錢！
Paper Trade 是使用虛擬資金進行的模擬交易：
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import uuid
class PaperTrade:
"""模擬交易系統"""
def __init__(self, initial_balance=100000):
self.initial_balance = initial_balance
self.balance = initial_balance
self.positions = {} # {symbol: {'side', 'size', 'entry_price', 'entry_time'}}
self.orders = {} # {order_id: order_info}
self.trade_history = []
self.pnl_history = []
self.market_data = {}
def add_market_data(self, symbol, data):
"""添加市場數據"""
self.market_data[symbol] = data
def get_current_price(self, symbol, timestamp=None):
"""獲取當前價格"""
if symbol not in self.market_data:
raise ValueError(f"No market data for {symbol}")
if timestamp is None:
return self.market_data[symbol]['close'].iloc[-1]
# 根據時間戳獲取歷史價格
data = self.market_data[symbol]
try:
return data.loc[data.index <= timestamp, 'close'].iloc[-1]
except IndexError:
return data['close'].iloc[0]
def place_order(self, symbol, side, size, order_type='market', price=None):
"""下單"""
order_id = str(uuid.uuid4())
timestamp = datetime.now()
order = {
'order_id': order_id,
'symbol': symbol,
'side': side, # 'buy' or 'sell'
'size': size,
'order_type': order_type,
'price': price,
'timestamp': timestamp,
'status': 'pending'
}
self.orders[order_id] = order
# 立即執行市價單
if order_type == 'market':
return self.execute_order(order_id)
return order_id
def execute_order(self, order_id):
"""執行訂單"""
if order_id not in self.orders:
return False, "Order not found"
order = self.orders[order_id]
symbol = order['symbol']
side = order['side']
size = order['size']
try:
# 獲取執行價格
if order['order_type'] == 'market':
execution_price = self.get_current_price(symbol)
else:
execution_price = order['price']
# 檢查餘額是否足夠
if side == 'buy':
required_balance = size * execution_price
if required_balance > self.balance:
order['status'] = 'rejected'
return False, "Insufficient balance"
# 更新持倉
self._update_position(symbol, side, size, execution_price, order['timestamp'])
# 更新餘額
if side == 'buy':
self.balance -= size * execution_price
else:
self.balance += size * execution_price
# 記錄交易
trade = {
'trade_id': str(uuid.uuid4()),
'order_id': order_id,
'symbol': symbol,
'side': side,
'size': size,
'price': execution_price,
'timestamp': order['timestamp'],
'pnl': self._calculate_trade_pnl(symbol, side, size, execution_price)
}
self.trade_history.append(trade)
order['status'] = 'filled'
order['execution_price'] = execution_price
return True, f"Order executed: {side} {size} {symbol} at {execution_price}"
except Exception as e:
order['status'] = 'error'
return False, f"Execution error: {str(e)}"
def _update_position(self, symbol, side, size, price, timestamp):
"""更新持倉"""
if symbol not in self.positions:
self.positions[symbol] = {
'side': side,
'size': size,
'entry_price': price,
'entry_time': timestamp
}
else:
position = self.positions[symbol]
if position['side'] == side:
# 加倉
total_size = position['size'] + size
weighted_price = (position['size'] * position['entry_price'] +
size * price) / total_size
position['size'] = total_size
position['entry_price'] = weighted_price
else:
# 反向交易
if size >= position['size']:
# 完全平倉或反向
remaining_size = size - position['size']
if remaining_size > 0:
position['side'] = side
position['size'] = remaining_size
position['entry_price'] = price
position['entry_time'] = timestamp
else:
del self.positions[symbol]
else:
# 部分平倉
position['size'] -= size
def _calculate_trade_pnl(self, symbol, side, size, price):
"""計算交易損益"""
if symbol not in self.positions:
return 0
position = self.positions[symbol]
if position['side'] != side:
# 平倉交易
if side == 'sell':
return size * (price - position['entry_price'])
else:
return size * (position['entry_price'] - price)
return 0 # 開倉交易無立即損益
def calculate_portfolio_value(self):
"""計算投資組合總值"""
total_value = self.balance
for symbol, position in self.positions.items():
try:
current_price = self.get_current_price(symbol)
position_value = position['size'] * current_price
if position['side'] == 'buy':
total_value += position_value
else:
# 做空倉位的價值計算
entry_value = position['size'] * position['entry_price']
pnl = position['size'] * (position['entry_price'] - current_price)
total_value += entry_value + pnl
except Exception:
continue
return total_value
def get_unrealized_pnl(self):
"""計算未實現損益"""
unrealized_pnl = {}
total_unrealized = 0
for symbol, position in self.positions.items():
try:
current_price = self.get_current_price(symbol)
if position['side'] == 'buy':
pnl = position['size'] * (current_price - position['entry_price'])
else:
pnl = position['size'] * (position['entry_price'] - current_price)
unrealized_pnl[symbol] = {
'position': position,
'current_price': current_price,
'unrealized_pnl': pnl,
'pnl_percentage': pnl / (position['size'] * position['entry_price'])
}
total_unrealized += pnl
except Exception:
continue
return unrealized_pnl, total_unrealized
def get_performance_metrics(self):
"""計算績效指標"""
if not self.trade_history:
return {}
trades = pd.DataFrame(self.trade_history)
# 計算總收益
total_pnl = trades['pnl'].sum()
# 計算勝率
winning_trades = trades[trades['pnl'] > 0]
losing_trades = trades[trades['pnl'] < 0]
win_rate = len(winning_trades) / len(trades) if len(trades) > 0 else 0
# 計算平均盈虧
avg_win = winning_trades['pnl'].mean() if len(winning_trades) > 0 else 0
avg_loss = losing_trades['pnl'].mean() if len(losing_trades) > 0 else 0
# 計算收益率
portfolio_value = self.calculate_portfolio_value()
total_return = (portfolio_value - self.initial_balance) / self.initial_balance
return {
'total_trades': len(trades),
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate': win_rate,
'total_pnl': total_pnl,
'avg_win': avg_win,
'avg_loss': avg_loss,
'profit_factor': abs(avg_win / avg_loss) if avg_loss != 0 else float('inf'),
'initial_balance': self.initial_balance,
'current_balance': self.balance,
'portfolio_value': portfolio_value,
'total_return': total_return
}
class MAStrategy:
"""移動平均策略 - Paper Trade 版本"""
def __init__(self, paper_trader, fast_period=20, slow_period=50):
self.paper_trader = paper_trader
self.fast_period = fast_period
self.slow_period = slow_period
self.last_signal = None
def calculate_signals(self, symbol, data):
"""計算交易信號"""
# 計算移動平均
data['fast_ma'] = data['close'].rolling(self.fast_period).mean()
data['slow_ma'] = data['close'].rolling(self.slow_period).mean()
# 生成信號
data['signal'] = 0
data.loc[data['fast_ma'] > data['slow_ma'], 'signal'] = 1 # 買入信號
data.loc[data['fast_ma'] < data['slow_ma'], 'signal'] = -1 # 賣出信號
return data
def execute_strategy(self, symbol, current_data):
"""執行策略"""
# 計算當前信號
signals = self.calculate_signals(symbol, current_data)
current_signal = signals['signal'].iloc[-1]
# 如果信號改變，執行交易
if current_signal != self.last_signal:
# 檢查當前持倉
current_position = self.paper_trader.positions.get(symbol)
if current_signal == 1: # 買入信號
if current_position is None or current_position['side'] == 'sell':
# 平掉空倉（如果有的話）
if current_position and current_position['side'] == 'sell':
self.paper_trader.place_order(
symbol, 'buy', current_position['size']
)
# 開多倉
trade_size = self.calculate_position_size(symbol)
self.paper_trader.place_order(symbol, 'buy', trade_size)
elif current_signal == -1: # 賣出信號
if current_position is None or current_position['side'] == 'buy':
# 平掉多倉（如果有的話）
if current_position and current_position['side'] == 'buy':
self.paper_trader.place_order(
symbol, 'sell', current_position['size']
)
# 開空倉
trade_size = self.calculate_position_size(symbol)
self.paper_trader.place_order(symbol, 'sell', trade_size)
self.last_signal = current_signal
def calculate_position_size(self, symbol):
"""計算倉位大小"""
# 簡單的固定比例倉位管理
current_price = self.paper_trader.get_current_price(symbol)
max_position_value = self.paper_trader.balance * 0.2 # 最大20%倉位
return max_position_value / current_price
# 策略回測示例
def run_strategy_simulation():
"""運行策略模擬"""
# 創建模擬交易器
paper_trader = PaperTrade(initial_balance=100000)
# 生成模擬數據
dates = pd.date_range('2024-01-01', '2024-12-31', freq='1D')
np.random.seed(42)
# 模擬 BTC 價格數據
returns = np.random.normal(0.001, 0.03, len(dates)) # 平均日收益0.1%，波動3%
prices = [50000] # 起始價格
for r in returns[1:]:
prices.append(prices[-1] * (1 + r))
btc_data = pd.DataFrame({
'close': prices,
'timestamp': dates
}, index=dates)
paper_trader.add_market_data('BTC', btc_data)
# 創建策略
strategy = MAStrategy(paper_trader)
# 執行回測
for i in range(50, len(btc_data)): # 從第50天開始（確保有足夠數據計算移動平均）
current_data = btc_data.iloc[:i+1]
strategy.execute_strategy('BTC', current_data)
# 獲取績效報告
metrics = paper_trader.get_performance_metrics()
print("模擬交易績效報告:")
print(f"總交易次數: {metrics['total_trades']}")
print(f"勝率: {metrics['win_rate']:.2%}")
print(f"總損益: ${metrics['total_pnl']:.2f}")
print(f"總收益率: {metrics['total_return']:.2%}")
print(f"盈虧比: {metrics['profit_factor']:.2f}")
return paper_trader, metrics
# 執行模擬
trader, results = run_strategy_simulation()
class RealisticPaperTrade(PaperTrade):
"""更真實的模擬交易系統"""
def __init__(self, initial_balance=100000, commission_rate=0.001, slippage_rate=0.0005):
super().__init__(initial_balance)
self.commission_rate = commission_rate
self.slippage_rate = slippage_rate
self.max_order_size_ratio = 0.1 # 最大單筆訂單佔市場成交量比例
def simulate_slippage(self, symbol, side, size, theoretical_price):
"""模擬滑價"""
# 根據訂單大小和方向計算滑價
market_impact = size * 0.00001 # 簡化的市場衝擊模型
if side == 'buy':
slippage = theoretical_price * (self.slippage_rate + market_impact)
execution_price = theoretical_price + slippage
else:
slippage = theoretical_price * (self.slippage_rate + market_impact)
execution_price = theoretical_price - slippage
return execution_price, slippage
def calculate_commission(self, symbol, size, price):
"""計算手續費"""
notional_value = size * price
commission = notional_value * self.commission_rate
return commission
def execute_order(self, order_id):
"""重寫訂單執行，加入現實因子"""
if order_id not in self.orders:
return False, "Order not found"
order = self.orders[order_id]
symbol = order['symbol']
side = order['side']
size = order['size']
try:
# 獲取理論執行價格
theoretical_price = self.get_current_price(symbol)
# 模擬滑價
execution_price, slippage = self.simulate_slippage(
symbol, side, size, theoretical_price
)
# 計算手續費
commission = self.calculate_commission(symbol, size, execution_price)
# 檢查餘額（包含手續費）
if side == 'buy':
required_balance = size * execution_price + commission
if required_balance > self.balance:
order['status'] = 'rejected'
return False, "Insufficient balance including commission"
# 執行交易
self._update_position(symbol, side, size, execution_price, order['timestamp'])
# 更新餘額（扣除手續費）
if side == 'buy':
self.balance -= (size * execution_price + commission)
else:
self.balance += (size * execution_price - commission)
# 記錄詳細交易資訊
trade = {
'trade_id': str(uuid.uuid4()),
'order_id': order_id,
'symbol': symbol,
'side': side,
'size': size,
'theoretical_price': theoretical_price,
'execution_price': execution_price,
'slippage': slippage,
'commission': commission,
'timestamp': order['timestamp'],
'pnl': self._calculate_trade_pnl(symbol, side, size, execution_price)
}
self.trade_history.append(trade)
order['status'] = 'filled'
order['execution_price'] = execution_price
order['slippage'] = slippage
order['commission'] = commission
return True, f"Order executed with slippage: {side} {size} {symbol} at {execution_price:.2f} (slippage: {slippage:.2f})"
except Exception as e:
order['status'] = 'error'
return False, f"Execution error: {str(e)}"
def get_detailed_performance_metrics(self):
"""獲取包含現實因子的詳細績效指標"""
base_metrics = self.get_performance_metrics()
if not self.trade_history:
return base_metrics
trades = pd.DataFrame(self.trade_history)
# 計算成本分析
total_commission = trades['commission'].sum()
total_slippage = trades['slippage'].sum()
total_trading_costs = total_commission + total_slippage
# 計算成本對績效的影響
gross_pnl = trades['pnl'].sum() + total_trading_costs
net_pnl = trades['pnl'].sum()
cost_impact = total_trading_costs / abs(gross_pnl) if gross_pnl != 0 else 0
base_metrics.update({
'total_commission': total_commission,
'total_slippage': total_slippage,
'total_trading_costs': total_trading_costs,
'gross_pnl': gross_pnl,
'net_pnl': net_pnl,
'cost_impact_percentage': cost_impact,
'average_commission_per_trade': total_commission / len(trades),
'average_slippage_per_trade': total_slippage / len(trades)
})
return base_metrics
# 現實模擬示例
realistic_trader = RealisticPaperTrade(
initial_balance=100000,
commission_rate=0.001, # 0.1% 手續費
slippage_rate=0.0005 # 0.05% 滑價
)
class ComprehensivePaperTrade:
"""綜合模擬交易環境"""
def __init__(self):
self.trading_hours = {
'crypto': {'start': 0, 'end': 24}, # 24/7
'forex': {'start': 0, 'end': 24}, # 週日晚到週五晚
'stocks': {'start': 9.5, 'end': 16} # 9:30-16:00
}
self.market_conditions = {
'normal': {'volatility_multiplier': 1.0, 'liquidity_multiplier': 1.0},
'high_volatility': {'volatility_multiplier': 2.0, 'liquidity_multiplier': 0.7},
'low_liquidity': {'volatility_multiplier': 1.2, 'liquidity_multiplier': 0.3},
'market_crash': {'volatility_multiplier': 5.0, 'liquidity_multiplier': 0.1}
}
def simulate_market_conditions(self, base_data, condition='normal'):
"""模擬不同市場條件"""
condition_params = self.market_conditions[condition]
volatility_mult = condition_params['volatility_multiplier']
liquidity_mult = condition_params['liquidity_multiplier']
# 調整價格波動
returns = base_data['close'].pct_change()
adjusted_returns = returns * volatility_mult
# 重新計算價格
adjusted_prices = [base_data['close'].iloc[0]]
for ret in adjusted_returns[1:]:
if not np.isnan(ret):
adjusted_prices.append(adjusted_prices[-1] * (1 + ret))
else:
adjusted_prices.append(adjusted_prices[-1])
# 調整流動性（影響滑價）
adjusted_data = base_data.copy()
adjusted_data['close'] = adjusted_prices
adjusted_data['liquidity_multiplier'] = liquidity_mult
return adjusted_data
def validate_strategy_robustness(self, strategy, test_scenarios):
"""測試策略在不同場景下的穩健性"""
results = {}
for scenario_name, scenario_data in test_scenarios.items():
# 為每個場景創建新的模擬交易器
trader = RealisticPaperTrade()
# 運行策略
strategy_instance = MAStrategy(trader)
for i in range(50, len(scenario_data)):
current_data = scenario_data.iloc[:i+1]
strategy_instance.execute_strategy('BTC', current_data)
# 記錄結果
metrics = trader.get_detailed_performance_metrics()
results[scenario_name] = metrics
return results
# 多場景測試
def comprehensive_strategy_test():
"""綜合策略測試"""
simulator = ComprehensivePaperTrade()
# 創建基礎數據
dates = pd.date_range('2024-01-01', '2024-12-31', freq='1D')
np.random.seed(42)
base_returns = np.random.normal(0.001, 0.02, len(dates))
base_prices = [50000]
for r in base_returns[1:]:
base_prices.append(base_prices[-1] * (1 + r))
base_data = pd.DataFrame({
'close': base_prices,
'timestamp': dates
}, index=dates)
# 創建不同市場場景
test_scenarios = {
'normal_market': simulator.simulate_market_conditions(base_data, 'normal'),
'high_volatility': simulator.simulate_market_conditions(base_data, 'high_volatility'),
'low_liquidity': simulator.simulate_market_conditions(base_data, 'low_liquidity'),
'market_crash': simulator.simulate_market_conditions(base_data, 'market_crash')
}
# 測試策略穩健性
results = simulator.validate_strategy_robustness(MAStrategy, test_scenarios)
# 分析結果
print("策略穩健性測試結果:")
print("-" * 50)
for scenario, metrics in results.items():
print(f"\n{scenario.upper()}:")
print(f"總收益率: {metrics.get('total_return', 0):.2%}")
print(f"勝率: {metrics.get('win_rate', 0):.2%}")
print(f"最大回撤: {metrics.get('max_drawdown', 0):.2%}")
print(f"交易成本影響: {metrics.get('cost_impact_percentage', 0):.2%}")
return results
# 執行測試
test_results = comprehensive_strategy_test()
class PsychologicalPaperTrade(RealisticPaperTrade):
"""包含心理因素的模擬交易"""
def __init__(self, initial_balance=100000):
super().__init__(initial_balance)
self.emotional_state = 'neutral' # 'confident', 'fearful', 'greedy', 'neutral'
self.consecutive_losses = 0
self.consecutive_wins = 0
self.stress_level = 0 # 0-10
def update_emotional_state(self):
"""更新情緒狀態"""
if self.consecutive_losses >= 3:
self.emotional_state = 'fearful'
self.stress_level = min(10, self.stress_level + 2)
elif self.consecutive_wins >= 3:
self.emotional_state = 'greedy'
self.stress_level = max(0, self.stress_level - 1)
else:
self.emotional_state = 'neutral'
self.stress_level = max(0, self.stress_level - 0.5)
def emotional_adjustment(self, original_size):
"""根據情緒狀態調整倉位大小"""
if self.emotional_state == 'fearful':
return original_size * 0.5 # 恐懼時減少倉位
elif self.emotional_state == 'greedy':
return original_size * 1.5 # 貪婪時增加倉位
else:
return original_size
def execute_order_with_emotion(self, order_id):
"""執行帶有情緒影響的訂單"""
if order_id not in self.orders:
return False, "Order not found"
order = self.orders[order_id]
# 情緒影響訂單執行
if self.emotional_state == 'fearful' and np.random.random() < 0.3:
order['status'] = 'cancelled'
return False, "Order cancelled due to fear"
# 調整訂單大小
original_size = order['size']
adjusted_size = self.emotional_adjustment(original_size)
order['size'] = adjusted_size
# 執行訂單
result = super().execute_order(order_id)
# 更新連續盈虧記錄
if result[0]: # 如果成功執行
trade = self.trade_history[-1]
if trade['pnl'] > 0:
self.consecutive_wins += 1
self.consecutive_losses = 0
elif trade['pnl'] < 0:
self.consecutive_losses += 1
self.consecutive_wins = 0
self.update_emotional_state()
return result
class PerformanceAnalyzer:
"""績效分析器"""
def __init__(self, paper_trader):
self.trader = paper_trader
def generate_comprehensive_report(self):
"""生成綜合報告"""
metrics = self.trader.get_detailed_performance_metrics()
report = {
'basic_metrics': self._analyze_basic_metrics(metrics),
'risk_metrics': self._analyze_risk_metrics(),
'trade_analysis': self._analyze_trades(),
'recommendations': self._generate_recommendations(metrics)
}
return report
def _analyze_basic_metrics(self, metrics):
"""分析基本指標"""
return {
'total_return': metrics.get('total_return', 0),
'win_rate': metrics.get('win_rate', 0),
'profit_factor': metrics.get('profit_factor', 0),
'average_trade': metrics.get('total_pnl', 0) / max(metrics.get('total_trades', 1), 1),
'trading_frequency': metrics.get('total_trades', 0) / 365 # 假設一年數據
}
def _analyze_risk_metrics(self):
"""分析風險指標"""
if not self.trader.trade_history:
return {}
trades = pd.DataFrame(self.trader.trade_history)
daily_pnl = trades.groupby(trades['timestamp'].dt.date)['pnl'].sum()
# 計算夏普比率
if len(daily_pnl) > 1:
sharpe_ratio = daily_pnl.mean() / daily_pnl.std() * np.sqrt(252)
else:
sharpe_ratio = 0
# 計算最大回撤
cumulative_pnl = daily_pnl.cumsum()
peak = cumulative_pnl.expanding().max()
drawdown = (cumulative_pnl - peak) / peak
max_drawdown = drawdown.min()
return {
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'volatility': daily_pnl.std() * np.sqrt(252),
'downside_deviation': daily_pnl[daily_pnl < 0].std() * np.sqrt(252)
}
def _analyze_trades(self):
"""分析交易模式"""
if not self.trader.trade_history:
return {}
trades = pd.DataFrame(self.trader.trade_history)
# 分析交易時間分佈
trades['hour'] = trades['timestamp'].dt.hour
hourly_pnl = trades.groupby('hour')['pnl'].sum()
# 分析持倉時間（需要配對買賣訂單）
# 這裡簡化處理
return {
'best_trading_hours': hourly_pnl.idxmax(),
'worst_trading_hours': hourly_pnl.idxmin(),
'total_commission_paid': trades['commission'].sum() if 'commission' in trades else 0,
'total_slippage_cost': trades['slippage'].sum() if 'slippage' in trades else 0
}
def _generate_recommendations(self, metrics):
"""生成改進建議"""
recommendations = []
if metrics.get('win_rate', 0) < 0.4:
recommendations.append("勝率偏低，考慮優化進場信號")
if metrics.get('profit_factor', 0) < 1.5:
recommendations.append("盈虧比需要改善，考慮調整停損停利設定")
if metrics.get('cost_impact_percentage', 0) > 0.3:
recommendations.append("交易成本過高，考慮降低交易頻率")
if not recommendations:
recommendations.append("策略表現良好，可考慮小額實盤測試")
return recommendations
# 生成完整報告
def generate_paper_trade_report(trader):
"""生成完整的模擬交易報告"""
analyzer = PerformanceAnalyzer(trader)
report = analyzer.generate_comprehensive_report()
print("=" * 60)
print("模擬交易完整報告")
print("=" * 60)
print("\n基本指標:")
for key, value in report['basic_metrics'].items():
print(f"{key}: {value:.4f}")
print("\n風險指標:")
for key, value in report['risk_metrics'].items():
print(f"{key}: {value:.4f}")
print("\n交易分析:")
for key, value in report['trade_analysis'].items():
print(f"{key}: {value}")
print("\n改進建議:")
for i, rec in enumerate(report['recommendations'], 1):
print(f"{i}. {rec}")
return report
今天我們深入學習了 Paper Trade（模擬交易），就像在試驗田裡先測試新品種一樣。模擬交易的重要價值：
Paper Trade 的優勢：
關鍵注意事項：
最佳實踐：
從模擬到實盤的過渡：
Paper Trade 就像農夫的試驗田，是通往成功交易的必經之路。明天我們將開始實作部分，把理論和模擬轉化為真正的交易系統！
下一篇：Day 26 - 範例實作說明
iThome鐵人賽
看影片追技術