Module 07: DataEngine – Market Data IngestionHow data is ingested, processed, and distributed from the market to strategies

1. Responsibilities of the Data Engine

flowchart LR
                        subgraph sources["🌍 Data Sources"]
                        ws["Exchange WebSockets"]
                        rest["REST APIs"]
                        file["Historical Data Files"]
                        end
                        subgraph dc["📡 DataClient Adapters"]
                        parse["Parse Raw Data"]
                        normalize["Standardize Format"]
                        end
                        subgraph de["📊 DataEngine"]
                        route["Route Distribution"]
                        aggregate["Bar Aggregation"]
                        validate["Data Validation"]
                        end
                        subgraph output["📤 Output"]
                        cache["💾 Cache"]
                        msgbus["📡 MessageBus"]
                        end
                        sources --> dc --> de --> output

2. The Subscription Mechanism

Strategies operating in an event-driven architecture do not “poll” for state. They “subscribe” to it. The Data Engine manages all subscription routing topographies:

# Subscribe to market data within the strategy's on_start() callback
def on_start(self):
    # Subscribe to Top-of-Book quotes — triggers on_quote_tick for each new quote update
    self.subscribe_quote_ticks(instrument_id)

    # Subscribe to atomic trade ticks
    self.subscribe_trade_ticks(instrument_id)

    # Subscribe to aggregated bars — e.g., 1-minute time bars
    self.subscribe_bars(BarType.from_str("BTCUSDT.BINANCE-1-MINUTE-LAST"))

    # Subscribe to L2 limit order book depth (10 levels deep)
    self.subscribe_order_book_at_interval(
        instrument_id,
        depth=10,
        interval_ms=1000,  # Synchronous snapshot interval every 1000ms
    )

3. Bar Aggregation — 18 Advanced Methods

Bars are not limited to simple time-based intervals. NautilusTrader supports a wide range of advanced aggregation methods:

# Bar type format schema: {instrument_id}-{size}-{aggregation_method}-{price_origin}
bar_type_1min  = BarType.from_str("BTCUSDT.BINANCE-1-MINUTE-LAST")
bar_type_tick  = BarType.from_str("BTCUSDT.BINANCE-100-TICK-LAST")
bar_type_vol   = BarType.from_str("BTCUSDT.BINANCE-1000-VOLUME-LAST")
bar_type_renko = BarType.from_str("BTCUSDT.BINANCE-50-RENKO-MID")

4. The DataClient Abstraction Layer

The DataClient acts as the adapter between external market data sources and the internal system. Each exchange or data provider implementation wraps its own network logic and emits normalized data into the DataEngine.

flowchart LR
                        subgraph clients["DataClient Implementations (Adapter Layer)"]
                        direction TB
                        binance_dc["Binance Client"]
                        ib_dc["IB Client"]
                        sim_dc["Backtest Client"]
                        end

                        de["📊 DataEngine<br/>Core Engine"]

                        clients ==> de

                        style de fill:#0f1419,stroke:#4fc3f7,stroke-width:2px
                        style clients fill:#0f1419,stroke:#ab47bc,stroke-width:1px

5. Interactive Exercise

def on_start(self):
    btc_id = InstrumentId.from_str("BTCUSDT.BINANCE")

    # 1. Real-time BBO quotes
    self.subscribe_quote_ticks(btc_id)

    # 2. Activity-weighted tick bars (500 trades)
    self.subscribe_bars(
        BarType.from_str("BTCUSDT.BINANCE-500-TICK-LAST")
    )

    # 3. 5-minute chronological bars
    self.subscribe_bars(
        BarType.from_str("BTCUSDT.BINANCE-5-MINUTE-LAST")
    )

6. Module Key Takeaways