模組 05B：數據管理與持久化從你的 CSV 到 NautilusTrader 能讀的格式

NautilusTrader 需要特定格式的數據才能運行回測。這個模組教你如何把各種來源的數據轉換、存儲、和載入到系統中。

一、數據管線全景

數據從你自己的檔案到回測引擎，需要經過以下步驟：

flowchart LR
                        subgraph source["📁 數據來源"]
                        direction TB
                        csv["CSV 檔案"]
                        api["交易所 API"]
                        vendor["數據供應商<br/>(Databento/Tardis)"]
                        end

                        subgraph transform["🔄 數據轉換"]
                        direction TB
                        wrangler["Wrangler<br/>數據整理器"]
                        loader["DataLoader<br/>數據載入器"]
                        end

                        subgraph storage["💾 數據存儲"]
                        direction TB
                        catalog["DataCatalog<br/>(Parquet 格式)"]
                        memory["直接載入記憶體"]
                        end

                        subgraph engine["⚙️ 回測引擎"]
                        direction TB
                        backtest["BacktestEngine"]
                        end

                        source ==> transform ==> storage ==> engine

二、Wrangler — 數據格式轉換器

從 CSV 載入 K 線數據

"""
從自己的 CSV 檔案載入 K 線數據到 NautilusTrader
"""
import pandas as pd
from nautilus_trader.persistence.wranglers import BarDataWrangler
from nautilus_trader.model.data import BarType
from nautilus_trader.model.identifiers import InstrumentId

# 1. 用 pandas 讀取你的 CSV
df = pd.read_csv("my_btc_data.csv")
# 你的 CSV 欄位可能是：timestamp, open, high, low, close, volume

# 2. 確保索引是時間戳
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")

# 3. 定義 BarType（什麼品種、什麼時間框架）
bar_type = BarType.from_str("BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL")
#                             品種                   時間框架  價格類型  來源

# 4. 用 Wrangler 轉換！
wrangler = BarDataWrangler(bar_type=bar_type, instrument=instrument)
bars = wrangler.process(df)  # 返回 list[Bar]

print(f"✅ 載入了 {len(bars)} 根 K 線")
print(f"第一根: {bars[0]}")
print(f"最後根: {bars[-1]}")

其它 Wrangler 種類

三、DataCatalog — 數據倉庫

flowchart LR
                        subgraph write["✍️ 寫入（只做一次）"]
                        direction TB
                        w1["CSV → Wrangler → Bars"]
                        w2["catalog.write_data(bars)"]
                        w1 --> w2
                        end

                        subgraph read["📖 讀取（每次回測）"]
                        direction TB
                        r1["catalog.bars()"]
                        r2["直接得到 list 對象"]
                        r1 --> r2
                        end

                        subgraph format["📦 Parquet 格式"]
                        direction TB
                        f1["二進制列式存儲"]
                        f2["壓縮比 5-10x"]
                        f3["支援時間範圍查詢"]
                        end

                        write --> format --> read

                        style format fill:#0f1419,stroke:#4fc3f7,stroke-width:2px

寫入數據到 Catalog

"""
把數據存入 DataCatalog（只需做一次）
"""
from nautilus_trader.persistence.catalog import ParquetDataCatalog

# 1. 指定 Catalog 的存儲目錄
catalog = ParquetDataCatalog("./my_catalog")

# 2. 寫入數據（bars 是上一步 Wrangler 產出的）
catalog.write_data(bars)
print(f"✅ 已寫入 {len(bars)} 根 K 線到 Catalog")

# 目錄結構會變成：
# ./my_catalog/
# └── bar/
#     └── BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL/
#         └── part-0.parquet      ← 二進制數據檔案

從 Catalog 讀取數據

"""
從 DataCatalog 讀取數據（每次回測都用這個）
"""
from nautilus_trader.persistence.catalog import ParquetDataCatalog

catalog = ParquetDataCatalog("./my_catalog")

# 讀取所有 K 線
bars = catalog.bars()

# 或者按時間範圍讀取（只讀 2024 年 1 月的數據）
from datetime import datetime
bars_jan = catalog.bars(
    start=datetime(2024, 1, 1),
    end=datetime(2024, 2, 1),
)

# 讀取特定品種
bars_btc = catalog.bars(
    bar_types=["BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL"],
)

print(f"📊 讀取了 {len(bars)} 根 K 線（速度：毫秒級！）")

四、K 線聚合 — 自定義時間框架

"""
在策略中訂閱自定義時間框架的 K 線
"""
class MyStrategy(Strategy):
    def on_start(self):
        # 訂閱 15 分鐘 K 線（引擎會自動從 1 分鐘 K 線聚合）
        bar_type_15m = BarType.from_str(
            "BTC-USDT-PERP.BINANCE-15-MINUTE-LAST-INTERNAL"
            #                                            ^^^^^^^^
            # INTERNAL = NautilusTrader 自己聚合
            # EXTERNAL = 由外部數據源提供
        )
        self.subscribe_bars(bar_type_15m)

        # 你甚至可以訂閱按「成交量」聚合的 K 線！
        bar_type_volume = BarType.from_str(
            "BTC-USDT-PERP.BINANCE-1000-VOLUME-LAST-INTERNAL"
            # 每累積 1000 單位成交量就生成一根 K 線
        )
        self.subscribe_bars(bar_type_volume)

    def on_bar(self, bar: Bar):
        # 當聚合完的 K 線到達時，這個回調被觸發
        self.log.info(f"收到 K 線: {bar}")

K 線聚合類型

五、自定義數據類型

如果你有自己的數據（例如：情緒指標、新聞事件），可以定義自定義數據類型在 NautilusTrader 中使用：

"""
定義一個自定義數據類型：市場情緒指數
"""
from nautilus_trader.model.custom import customdataclass

# 用 @customdataclass 裝飾器定義自定義數據
@customdataclass
class SentimentData:
    instrument_id: str   # 品種 ID
    score: float         # 情緒分數 (-1.0 到 1.0)
    source: str          # 來源（Twitter, Reddit 等）

# 在策略中訂閱和使用
class MyStrategy(Strategy):
    def on_start(self):
        # 訂閱自定義數據
        self.subscribe_data(
            data_type=SentimentData,
            metadata={"instrument_id": "BTC-USDT-PERP.BINANCE"},
        )

    def on_data(self, data):
        if isinstance(data, SentimentData):
            if data.score > 0.8:
                self.log.info("🟢 市場情緒極度樂觀！")
            elif data.score < -0.8:
                self.log.info("🔴 市場情緒極度悲觀！")

六、序列化格式

NautilusTrader 支援多種序列化格式，用於不同場景：

七、互動練習

八、本模組重點回顧