Skip to main content

MarketDataSource

brokapi_core::traits

Trait MarketDataSource 

Source 
pub trait MarketDataSource: Send + Sync {
    // Required methods
    fn last_trades<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 InstrumentId,
        limit: u32,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<MarketTrade>, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn subscribe_market_trades<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 InstrumentId,
        opts: SubscribeOpts,
    ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn subscribe_quotes<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 InstrumentId,
        opts: SubscribeOpts,
    ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn candles<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 InstrumentId,
        tf: Timeframe,
        range: TimeRange,
    ) -> Pin<Box<dyn Future<Output = Result<Vec<Candle>, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn subscribe_candles<'life0, 'life1, 'async_trait>(
        &'life0 self,
        id: &'life1 InstrumentId,
        tf: Timeframe,
        opts: SubscribeOpts,
    ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;

    // Provided method
    fn candles_max_span(&self, tf: Timeframe) -> Option<Duration> { ... }
}
Expand description

Public trades, quotes and candles: history reads and streams.

Reads (last_trades, candles) answer directly. Streams (subscribe_market_trades, subscribe_quotes, subscribe_candles) return a SubscriptionId and send Event::MarketTrade, Event::Quote and Event::Candle into the channel given at start. Stop a stream with OrderBookSource::unsubscribe, which takes any subscription of the connector. Your own fills are not here, see ExecutionSource.

§Examples

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::{
    Event, InstrumentId, InstrumentKey, MarketDataSource, NativeId, ShutdownMode, SubscribeOpts,
    TimeRange, Timeframe, UnixNanos,
};
use brokapi_runtime::BrokerLimiter;

let (events_tx, mut events) = tokio::sync::mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);

// The last 2 deals of the tape, oldest first (BCS).
let trades = md.last_trades(&sber, 2).await?;
assert!(!trades.is_empty() && trades.len() <= 2);

// One day of 1-minute candles. The range is in nanoseconds since the epoch.
let day = TimeRange::new(
    UnixNanos::from_nanos(1_792_368_000_000_000_000), // 2026-10-19 00:00 UTC
    UnixNanos::from_nanos(1_792_454_400_000_000_000), // 2026-10-20 00:00 UTC
);
let bars = md.candles(&sber, Timeframe::M1, day).await?;
assert!(bars.iter().all(|c| c.low <= c.high));

// A stream of quotes: the best bid and ask, the latest one wins if we read slowly.
let sub = md.subscribe_quotes(&sber, SubscribeOpts::conflate()).await?;
let quote = tokio::time::timeout(Duration::from_secs(8), async {
    while let Some(env) = events.recv().await {
        if let (Some(s), Event::Quote(q)) = (env.sub, env.payload) {
            if s == sub {
                return Some(q);
            }
        }
    }
    None
})
.await?
.expect("a quote arrived");
assert!(quote.bid.is_some() || quote.ask.is_some());

md.shutdown(ShutdownMode::Immediate).await;

Required Methods§

Source

fn last_trades<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, limit: u32, ) -> Pin<Box<dyn Future<Output = Result<Vec<MarketTrade>, Error>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Reads the newest limit deals of the public tape.

How far back the tape goes depends on the broker (BCS: the current session, T-Bank: the last hour). The BCS connector returns the deals oldest first and drops a record whose side is unknown. You may get fewer than limit.

§Errors
§Examples
use brokapi_core::MarketDataSource;
let trades = md.last_trades(&sber, 5).await?;
for t in &trades {
    assert!(t.price.is_positive() && t.qty.is_positive());
}
Source

fn subscribe_market_trades<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, opts: SubscribeOpts, ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Starts a stream of the public deals of one instrument.

Deals arrive as Event::MarketTrade tagged with the returned id. Every deal is an event of its own, so Backpressure::Conflate would lose deals: use Backpressure::DropNewest if losing some is acceptable, or Backpressure::Lossless if not (an overflow then sends Event::Overflow).

§Errors
§Examples
use std::time::Duration;
use brokapi_core::{Backpressure, Event, MarketDataSource, SubscribeOpts};
// Keep up to 256 deals while we are slow, drop the newest ones beyond that.
let opts = SubscribeOpts { backpressure: Backpressure::DropNewest { cap: 256 }, depth: None };
let sub = md.subscribe_market_trades(&sber, opts).await?;
let env = tokio::time::timeout(Duration::from_secs(8), events.recv()).await?.expect("channel is open");
assert_eq!(env.sub, Some(sub));
assert!(matches!(env.payload, Event::MarketTrade(_)));
Source

fn subscribe_quotes<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, opts: SubscribeOpts, ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Starts a stream of the best bid and ask of one instrument.

Quotes arrive as Event::Quote tagged with the returned id. Each quote replaces the previous one, so Backpressure::Conflate is the right policy.

§Errors
§Examples

See the example of the trait, which receives a quote.

Source

fn candles<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, tf: Timeframe, range: TimeRange, ) -> Pin<Box<dyn Future<Output = Result<Vec<Candle>, Error>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Reads the bars of tf that fall in range.

A bar has open_ts and close_ts, and a price is a Price. Whether the ends of range are inclusive and how far back the broker serves is decided by the broker.

§Errors
§Examples
use brokapi_core::{Error, MarketDataSource, TimeRange, Timeframe, UnixNanos};
let day = TimeRange::new(
    UnixNanos::from_nanos(1_792_368_000_000_000_000), // 2026-10-19 00:00 UTC
    UnixNanos::from_nanos(1_792_454_400_000_000_000), // 2026-10-20 00:00 UTC
);
let bars = md.candles(&sber, Timeframe::M5, day).await?;
assert!(bars.iter().all(|c| c.timeframe == Timeframe::M5));

// BCS has no 10-minute bars. The answer comes at once, nothing is sent to the broker.
match md.candles(&sber, Timeframe::M10, day).await {
    Err(Error::Unsupported { capability }) => assert_eq!(capability, "candles_m10"),
    other => panic!("expected Unsupported, got {other:?}"),
}
Source

fn subscribe_candles<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, tf: Timeframe, opts: SubscribeOpts, ) -> Pin<Box<dyn Future<Output = Result<SubscriptionId, Error>> + Send + 'async_trait>>
where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Starts a stream of the latest bar of tf for one instrument.

Bars arrive as Event::Candle tagged with the returned id. Use Backpressure::Conflate: the latest bar replaces older ones.

§Errors
§Examples
use std::time::Duration;
use brokapi_core::{Event, MarketDataSource, SubscribeOpts, Timeframe};
let sub = md.subscribe_candles(&sber, Timeframe::M1, SubscribeOpts::conflate()).await?;
let env = tokio::time::timeout(Duration::from_secs(8), events.recv()).await?.expect("channel is open");
assert_eq!(env.sub, Some(sub));
let Event::Candle(bar) = env.payload else { panic!("a candle subscription sends candles") };
assert_eq!(bar.timeframe, Timeframe::M1);

Provided Methods§

Source

fn candles_max_span(&self, tf: Timeframe) -> Option<Duration>

The longest span of time one candles request may cover for tf, or None when the connector does not know a limit.

Brokers cap one history request by the number of bars or by the length of the span, and a request over the cap is refused (Error::Rejected). A long history is therefore read in windows of at most this span. Do not slice by hand: brokapi_runtime::history does it, waits out Error::RateLimited and joins the windows. The value comes from the broker’s documentation, not from a live check. Pure function, no network call.

The default is None (“unknown”): the caller then chooses a window itself.

Dyn Compatibility§

This trait is dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors§