Lab 01：從零到回測動手實作：建立你的第一個完整的 NautilusTrader 專案

目標

在這個 Lab 中，你將從一個空白資料夾開始，一步步搭建出一個可以運行的 EMA 交叉策略回測系統。你將學會：

1. 專案結構的最佳實踐
2. 準備 CSV 數據並轉換為 Parquet
3. 編寫 DataCatalog 配置
4. 編寫 EMA Cross 策略
5. 配置 BacktestEngine 並運行回測
6. 生成 Tearsheet 分析結果

步驟 1：建立專案結構

在你的工作目錄（例如 my_quant_project）中，建立以下文件夾和文件：

my_quant_project/
├── data/               # 存放 CSV 原始數據
├── catalog/            # 存放 Parquet 數據
├── strategies/         # 存放策略代碼
│   ├── __init__.py
│   └── ema_cross.py    # 我們的策略文件
├── config.py           # 全局配置
└── run_backtest.py     # 回測啟動腳本

步驟 2：準備數據 (CSV → Parquet)

NautilusTrader 推薦使用 Partquet 格式。我們寫一個腳本把 CSV 轉成 Parquet。

假設你有一個 data/ETHUSDT.csv，格式如下（如果是 Binance 數據）：

timestamp,open,high,low,close,volume,...
1609459200000,736.42,739.00,736.00,737.95,123.45,...

創建 convert_data.py：

import pandas as pd
from nautilus_trader.model.data import Bar, BarType
from nautilus_trader.model.enums import BarAggregation, PriceType
from nautilus_trader.model.instruments import InstrumentId
from nautilus_trader.persistence.catalog import ParquetDataCatalog

def convert_csv_to_parquet():
    # 1. 讀取 CSV
    df = pd.read_csv("data/ETHUSDT.csv")
    
    # 2. 為每一行創建 Bar 對象 (這一步比較慢，但只需做一次)
    bars = []
    instrument_id = InstrumentId.from_str("ETHUSDT.BINANCE")
    bar_type = BarType.from_str("ETHUSDT.BINANCE-1-MINUTE-LAST-EXTERNAL")
    
    for _, row in df.iterrows():
        bar = Bar(
            bar_type=bar_type,
            ts_event=int(row["timestamp"] * 1_000_000), # 毫秒轉納秒
            ts_init=int(row["timestamp"] * 1_000_000),
            open=float(row["open"]),
            high=float(row["high"]),
            low=float(row["low"]),
            close=float(row["close"]),
            volume=float(row["volume"]),
        )
        bars.append(bar)

    # 3. 寫入 Catalog
    catalog = ParquetDataCatalog("./catalog")
    catalog.write_data(bars)
    print("數據轉換完成！")

if __name__ == "__main__":
    convert_csv_to_parquet()

步驟 3：編寫策略 (strategies/ema_cross.py)

from decimal import Decimal
from nautilus_trader.trading.strategy import Strategy
from nautilus_trader.config import StrategyConfig
from nautilus_trader.indicators.average.ema import ExponentialMovingAverage
from nautilus_trader.model.enums import OrderSide

class EmaCrossConfig(StrategyConfig):
    instrument_id: str
    fast_period: int = 10
    slow_period: int = 20
    trade_size: str = "0.01"

class EmaCrossStrategy(Strategy):
    def __init__(self, config: EmaCrossConfig):
        super().__init__(config)
        self.instrument_id = config.instrument_id
        
        # 定義指標
        self.fast_ema = ExponentialMovingAverage(config.fast_period)
        self.slow_ema = ExponentialMovingAverage(config.slow_period)

    def on_bar(self, bar):
        # 更新指標
        self.fast_ema.update(bar.close)
        self.slow_ema.update(bar.close)

        # 確保指標已準備好
        if not self.fast_ema.initialized or not self.slow_ema.initialized:
            return

        # 獲取當前持倉
        pos = self.portfolio.position(self.instrument_id)
        
        # 交易邏輯
        if self.fast_ema.value > self.slow_ema.value:
            # 黃金交叉：買入
            if not pos:
                self.buy()
        elif self.fast_ema.value < self.slow_ema.value:
            # 死亡交叉：賣出
            if pos:
                self.sell_all()

    def buy(self):
        order = self.order_factory.market(
            instrument_id=self.instrument_id,
            order_side=OrderSide.BUY,
            quantity=self.config.trade_size,
        )
        self.submit_order(order)

    def sell_all(self):
        self.close_all_positions(self.instrument_id)

步驟 4：運行回測 (run_backtest.py)

from decimal import Decimal
from nautilus_trader.backtest.engine import BacktestEngine, BacktestEngineConfig
from nautilus_trader.model.instruments import InstrumentId
from nautilus_trader.model.currencies import Currency
from nautilus_trader.model.objects import Money
from nautilus_trader.model.enums import AccountType, OmsType

from strategies.ema_cross import EmaCrossStrategy, EmaCrossConfig

#配置回測引擎
config = BacktestEngineConfig(
    strategies=[
        EmaCrossConfig(
            instrument_id="ETHUSDT.BINANCE",
            fast_period=10,
            slow_period=20,
            trade_size="0.1",
        )
    ]
)
engine = BacktestEngine(config=config)

# 1. 配置交易所和帳戶
engine.add_venue(
    venue="BINANCE",
    oms_type=OmsType.NETTING,
    account_type=AccountType.MARGIN,
    starting_balances=[Money(10_000, Currency.from_str("USDT"))],
)

# 2. 載入數據 (從我們剛才建立的 Catalog)
engine.add_data_from_catalog("./catalog")

# 3. 運行回測
engine.run()

# 4. 輸出結果
engine.trader.generate_account_report("BINANCE")
print("回測完成！")

# 5. 生成 Tearsheet (可選，需安裝 quantstats)
# 這裡展示簡易結果
portfolio = engine.trader.portfolio
print(f"最終權益: {portfolio.account('BINANCE').balance_total(Currency.from_str('USDT'))}")

作業