Module 05B: Data Management and PersistenceFrom Raw CSV to Native Data Structures

NautilusTrader expects market data in well-defined native types before you can run reliable backtests. This module walks through the end-to-end pipeline for loading, transforming, and storing high-fidelity data.

1. Global Data Pipeline Architecture

From flat files to the backtesting engine, data propagates through the following stages:

flowchart LR
                        subgraph source["📁 Data Sources"]
                        direction TB
                        csv["CSV Files"]
                        api["Exchange API"]
                        vendor["Data Vendors<br />(Databento/Tardis)"]
                        end

                        subgraph transform["🔄 Data Transformation"]
                        direction TB
                        wrangler["Wrangler<br />Data Normalizer"]
                        loader["DataLoader<br />Data Loader"]
                        end

                        subgraph storage["💾 Data Storage"]
                        direction TB
                        catalog["DataCatalog<br />(Parquet Format)"]
                        memory["In-Memory Load"]
                        end

                        subgraph engine["⚙️ Backtest Engine"]
                        direction TB
                        backtest["BacktestEngine"]
                        end

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

2. Wrangler — The Data Format Converter

Transforming CSV to Core Bar Objects

"""
Loading historical CSV data into NautilusTrader native objects.
"""
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. Read CSV using pandas
df = pd.read_csv("my_btc_data.csv")
# Expected columns: timestamp, open, high, low, close, volume

# 2. Ensure timestamp column is datetime indexed
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.set_index("timestamp")

# 3. Define the BarType (instrument, timeframe, price type, origin)
bar_type = BarType.from_str("BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL")
#                             Instrument             Timeframe PriceType Origin

# 4. Process using Wrangler
wrangler = BarDataWrangler(bar_type=bar_type, instrument=instrument)
bars = wrangler.process(df)  # Returns list[Bar]

print(f"✅ Loaded {len(bars)} bars")
print(f"First bar: {bars[0]}")
print(f"Last bar: {bars[-1]}")

Wrangler Types

3. DataCatalog — The Data Warehouse

flowchart LR
                        subgraph write["✍️ Write (One-time)"]
                        direction TB
                        w1["CSV → Wrangler → Bars"]
                        w2["catalog.write_data(bars)"]
                        w1 --> w2
                        end

                        subgraph read["📖 Read (Per Backtest)"]
                        direction TB
                        r1["catalog.bars()"]
                        r2["Returns list of objects"]
                        r1 --> r2
                        end

                        subgraph format["📦 Parquet Format"]
                        direction TB
                        f1["Binary Columnar Storage"]
                        f2["5-10x Compression"]
                        f3["Supports Time-range Queries"]
                        end

                        write --> format --> read

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

Persisting Wrangler Output into the Catalog

"""
Persisting data to DataCatalog (One-time operation)
"""
from nautilus_trader.persistence.catalog import ParquetDataCatalog

# 1. Define Catalog storage directory
catalog = ParquetDataCatalog("./my_catalog")

# 2. Write data (bars processed from Wrangler)
catalog.write_data(bars)
print(f"✅ Successfully written {len(bars)} bars to Catalog")

# Directory structure generated:
# ./my_catalog/
# └── bar/
#     └── BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL/
#         └── part-0.parquet      ← Binary data file

Querying Data from the Catalog

"""
Reading data from DataCatalog (Executed per backtest run)
"""
from nautilus_trader.persistence.catalog import ParquetDataCatalog

catalog = ParquetDataCatalog("./my_catalog")

# Retrieve all bars
bars = catalog.bars()

# Query by time range (e.g., only January 2024 data)
from datetime import datetime
bars_jan = catalog.bars(
    start=datetime(2024, 1, 1),
    end=datetime(2024, 2, 1),
)

# Query by specific instrument
bars_btc = catalog.bars(
    bar_types=["BTC-USDT-PERP.BINANCE-1-MINUTE-LAST-EXTERNAL"],
)

print(f"📊 Retrieved {len(bars)} bars in milliseconds")

4. Bar Aggregation — Custom Timeframes

"""
Subscribing to custom timeframe bars within a strategy
"""
class MyStrategy(Strategy):
    def on_start(self):
        # Subscribe to 15-minute bars (Engine aggregates autonomously from 1-minute bars)
        bar_type_15m = BarType.from_str(
            "BTC-USDT-PERP.BINANCE-15-MINUTE-LAST-INTERNAL"
            #                                            ^^^^^^^^
            # INTERNAL = Aggregated internally by NautilusTrader
            # EXTERNAL = Provided by an external data source
        )
        self.subscribe_bars(bar_type_15m)

        # You can even subscribe to volume-aggregated bars!
        bar_type_volume = BarType.from_str(
            "BTC-USDT-PERP.BINANCE-1000-VOLUME-LAST-INTERNAL"
            # Generates one bar per 1,000 units of accumulated volume
        )
        self.subscribe_bars(bar_type_volume)

    def on_bar(self, bar: Bar):
        # Triggered when the aggregated bar is generated
        self.log.info(f"Received bar: {bar}")

Bar Aggregation Types

5. Custom Data Types

If you use alternative data (e.g., sentiment indices, macroeconomic events), you can define custom data types to integrate seamlessly into NautilusTrader:

"""
Define a custom data type: Market Sentiment Index
"""
from nautilus_trader.model.custom import customdataclass

# Define custom data using the @customdataclass decorator
@customdataclass
class SentimentData:
    instrument_id: str   # Target instrument ID
    score: float         # Sentiment score (-1.0 to 1.0)
    source: str          # Data source (Twitter, Reddit, etc.)

# Subscribe to and use inside a strategy
class MyStrategy(Strategy):
    def on_start(self):
        # Subscribe to custom data
        self.subscribe_data(
            data_type=SentimentData,
            metadata={"instrument_id": "BTC-USDT-PERP.BINANCE"},
        )

    def on_data(self, data):
        # Process custom data updates
        if isinstance(data, SentimentData):
            if data.score > 0.8:
                self.log.info("🟢 Extreme bullish sentiment detected!")
            elif data.score < -0.8:
                self.log.info("🔴 Extreme bearish sentiment detected!")

6. Serialization Formats

NautilusTrader supports multiple serialization formats optimized for specific architectural requirements:

7. Interactive Exercises

8. Module Highlights