Struct DatabentoLive 

pub struct DatabentoLive<K> { /* private fields */ }

Live streaming client for Databento.

Wraps LiveClient for real-time market data streaming. Each instance connects to a single dataset but can subscribe to multiple symbols within that dataset.

Connection Limits

• Standard tier: 10 concurrent connections per dataset
• Enterprise tier: 50 concurrent connections per dataset

Consolidate symbol subscriptions within one DatabentoLive instance per dataset rather than creating multiple instances for the same dataset.

Performance

The instrument key is cloned for each record. For high-frequency data, use Arc<K> to avoid per-record heap allocations:

use std::sync::Arc;

let instruments: HashMap<String, Arc<String>> = [
    ("ESM5".to_string(), Arc::new("ES-front".to_string())),
].into_iter().collect();

Implementations

impl<K> DatabentoLive<K>

pub async fn from_env( dataset: &str, exchange: ExchangeId, instruments: HashMap<String, K>, ) -> Result<Self, DataError>

Create a new live client using API key from environment.

Reads DATABENTO_API_KEY from environment variables.

Arguments

• dataset - Databento dataset identifier (e.g., “GLBX.MDP3”, “XNAS.ITCH”)
• exchange - ExchangeId to tag events with (should match dataset)
• instruments - Map from Databento symbol strings to user’s instrument keys

Errors

Returns error if DATABENTO_API_KEY is not set or client construction fails.

pub async fn new( api_key: &str, dataset: &str, exchange: ExchangeId, instruments: HashMap<String, K>, ) -> Result<Self, DataError>

Create a new live client with an explicit API key.

Errors

Returns error if client construction fails.

pub async fn subscribe( &mut self, symbols: &[&str], schema: Schema, ) -> Result<(), DataError>

Subscribe to symbols with a specific schema.

Can be called multiple times before start() to add more subscriptions. All subscriptions are multiplexed over the same connection.

Arguments

• symbols - Symbol identifiers (e.g., ["ESM5", "NQM5"])
• schema - Data schema to subscribe to (e.g., Schema::Trades, Schema::Mbp1)

OHLCV note

This is the low-level escape hatch: it subscribes to any Schema as-is and performs no interval validation. Subscribing to an OHLCV schema here (e.g. Schema::Ohlcv1H/Schema::Ohlcv1D) bypasses the Sec1/Min1 live-only check in subscribe_candles, and any resulting bars whose rtype maps to a CandleInterval will be emitted as DataKind::Candle. Prefer subscribe_candles for candles unless you deliberately want that lower-level behaviour.

Errors

Returns error if subscription request fails.

pub async fn subscribe_candles( &mut self, symbols: &[&str], interval: CandleInterval, ) -> Result<(), DataError>

Subscribe to OHLCV candles for symbols at the given interval.

Validates the interval and maps it to the Databento OHLCV schema before subscribing. The resulting stream emits DataKind::Candle events whose interval is derived per-record from each bar’s rtype, so a single connection may safely carry multiple OHLCV intervals.

Live interval scope

Only Sec1/Min1 are accepted live. Databento’s live gateway does not reliably stream the larger Hour1/Day1 bars, so those are historical-only here — use DatabentoHistorical::fetch_candles. They are rejected at subscribe time (observable failure) rather than yielding a silently empty stream.

Errors

Returns DataError::UnsupportedInterval for any interval other than Sec1/Min1, or DataError::Databento if the subscription request fails.

pub fn client(&self) -> &LiveClient

Returns a reference to the underlying client for advanced use cases.

pub fn client_mut(&mut self) -> &mut LiveClient

Returns a mutable reference to the underlying client.

impl<K: Clone + Send + 'static> DatabentoLive<K>

pub async fn start( self, ) -> Result<impl Stream<Item = Result<MarketEvent<K, DataKind>, DataError>>, DataError>

Start streaming and return an owned stream of market events.

Consumes self because LiveClient requires mutable access for polling. The stream will emit events until the connection closes or an error occurs.

Returns

A stream of MarketEvent<K, DataKind> where DataKind is one of Trade, OrderBookL1, or Candle depending on the subscribed schema(s).

Records for symbols not in the instruments map are skipped with a warning.

Errors

Returns error if starting the stream fails. Stream items may also contain errors for individual record processing failures.

Trait Implementations

impl<K: Debug> Debug for DatabentoLive<K>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.