Struct MarketDataConnector 

pub struct MarketDataConnector { /* private fields */ }

BCS market data connector: order books, trades, quotes, candles and the instrument directory.

One connector per broker is enough: it is not tied to an account. It implements [OrderBookSource], [MarketDataSource] and [InstrumentSource]. Commands return their answer directly. Subscriptions deliver [Envelope]s into the channel given to start.

Examples

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::{InstrumentId, InstrumentKey, NativeId, OrderBook, OrderBookSource};
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let (events_tx, _events) = 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() },
);
let book = md.order_book(&sber, 10).await?;
let OrderBook::Snapshot { bids, asks, .. } = book else { panic!("REST answers a snapshot") };
assert!(!bids.is_empty() && !asks.is_empty());

Implementations

impl MarketDataConnector

pub async fn start( cfg: MarketDataConfig, limiter: Arc<BrokerLimiter>, out: Sender<Envelope>, ) -> Result<MarketDataConnector, Error>

Starts the connector: gets the first access token and opens the market data WebSocket.

The refresh token in cfg is exchanged for an access token before anything else, so a wrong token fails here and not on the first request. This step does not spend rate limit quota. The WebSocket session is started in the background. It reconnects on its own and restores the subscriptions after a reconnect. A reconnect voids the order books: the stream tells the bot with Event::Resync before the new snapshot.

limiter is shared by value of Arc: give the same limiter to every connector of one BCS account, because the limits belong to the token. out receives the envelopes of all subscriptions. Its capacity is the bot’s choice. The bot must read it, because a slow reader triggers the backpressure policy of each subscription.

Errors

• [Error::Config] when the HTTP or WebSocket client cannot be built from cfg.
• [Error::Transport] when the token refresh fails: no connection, a timeout, or the auth server refuses the refresh token. A refused or expired refresh token is a permanent failure, it will not heal on retry (exact auth server answers: not verified against the live API).
• [Error::RateLimited] when the auth server answers HTTP 429.

Examples

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
assert_eq!(md.caps().max_order_book_depth, 20);
md.shutdown(ShutdownMode::Immediate).await;

A broker that cannot be reached is a [Error::Transport], and nothing is left running:

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::Error;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;
use brokapi_bcs_protocol::config::BcsConfig;

// Port 1 of the local host: nothing listens there.
let cfg = BcsConfig::for_mock("any-token", "http://127.0.0.1:1");
let (events_tx, _events) = mpsc::channel(16);
let limiter = BrokerLimiter::new(&default_limits());
let started = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await;
assert!(matches!(started, Err(Error::Transport(_))));

pub fn raw(&self) -> BcsRaw

Returns a handle for any other method of the BCS API, with this connector’s token, endpoint and rate limiter.

The handle is cheap to clone and stays usable while the connector lives. Every call through it spends quota by the operation name you give. See BcsRaw for the rules of a call.

Examples

A method the shared traits do not wrap, here the quotes snapshot:

use brokapi_bcs_protocol::http::models::{QuoteInstrument, QuotesRequest};
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;

let quotes = md
    .raw()
    .call("GetQuotes", |http| async move {
        http.get_quotes(&QuotesRequest { instruments: vec![QuoteInstrument::new("SBER", "TQBR")] }).await
    })
    .await?;
assert!(!quotes.records.is_empty());
md.shutdown(ShutdownMode::Immediate).await;

pub fn caps(&self) -> MarketDataCaps

Describes what this connector can do, without any request.

The answer is fixed for BCS: order books up to 20 levels, always as snapshots (incremental_order_book is false), and streams for order books, market trades, quotes and candles. A deeper depth in a request is cut to the maximum, it is not an error.

Examples

use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::ShutdownMode;
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
let caps = md.caps();
assert_eq!(caps.max_order_book_depth, 20);
assert!(!caps.incremental_order_book);
assert!(caps.streaming.order_book && caps.streaming.candles);

pub async fn shutdown(self, mode: ShutdownMode) -> ShutdownReport

Stops the connector and reports what was cleaned up. Consumes the connector.

• [ShutdownMode::Immediate]: no unsubscribe messages are sent, the WebSocket just goes away.
• [ShutdownMode::Graceful]: every subscription is unsubscribed on the WebSocket first. The counts are in the report (unsubscribed, unsubscribe_failed). Then the events that still wait in the subscription buffers get time to reach the bot’s channel, until deadline.

After the call the event pump has stopped, and no more envelopes arrive. Market data holds no orders, so inflight_abandoned is always empty here.

Examples

A graceful shutdown with one live subscription:

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::{Backpressure, InstrumentId, InstrumentKey, NativeId, OrderBookSource, ShutdownMode, SubscribeOpts};
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

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() },
);
md.subscribe_order_book(&sber, SubscribeOpts { backpressure: Backpressure::Conflate, depth: Some(10) }).await?;

let report = md.shutdown(ShutdownMode::Graceful { deadline: Duration::from_secs(1) }).await;
assert_eq!(report.unsubscribed + report.unsubscribe_failed, 1);
assert!(report.inflight_abandoned.is_empty());

Trait Implementations

impl Connector for MarketDataConnector

type Caps = MarketDataCaps

The type that describes what this connector can do.

fn caps(&self) -> MarketDataCaps

What this connector can do. Read it before you use a feature that not every broker has.

fn shutdown<'async_trait>( self, mode: ShutdownMode, ) -> Pin<Box<dyn Future<Output = ShutdownReport> + Send + 'async_trait>>

where MarketDataConnector: 'async_trait,

Stops the connector: closes its streams, gives up commands that have no answer yet (they become [Error::Unknown], see [ShutdownReport]) and ends its tasks. Takes the connector by value, so it cannot be used afterwards.

impl InstrumentSource for MarketDataConnector

fn schedule<'life0, 'life1, 'async_trait>( &'life0 self, exchange: &'life1 str, range: TimeRange, ) -> Pin<Box<dyn Future<Output = Result<Schedule, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

GET /trading-schedule/daily-schedule needs the instrument (ticker and classCode are both required), so BCS schedules are per instrument, like Finam’s: exchange is "TICKER@CLASSCODE" ("SBER@TQBR"). Anything else is a Config error. The answer is for one day, today: the intervals are put on the day of range.from (an assumption, see map/schedule.rs).

fn find<'life0, 'async_trait>( &'life0 self, q: InstrumentQuery, ) -> Pin<Box<dyn Future<Output = Result<Vec<Instrument>, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, MarketDataConnector: 'async_trait,

Looks instruments up in the directory.

fn get<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, ) -> Pin<Box<dyn Future<Output = Result<Instrument, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Reads one directory entry.

fn trading_params<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, ) -> Pin<Box<dyn Future<Output = Result<TradingParams, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Reads lot size, price step and the other numbers needed to size and price an order.

fn trading_status<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, ) -> Pin<Box<dyn Future<Output = Result<TradingStatus, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Reads whether the instrument can be traded now.

impl MarketDataSource for MarketDataConnector

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 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

POST /last-trades: the tape of the current session. The API has no limit, so the newest limit trades are kept here (oldest first). A record without a known side is dropped.

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

One request returns at most 1440 bars (BCS documentation of candles-chart), so the span is 1440 bars of the timeframe. Not checked against the live API.

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 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Starts a stream of the public deals of one instrument.

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 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

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

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 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Reads the bars of tf that fall in range.

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 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

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

impl OrderBookSource for MarketDataConnector

fn order_book<'life0, 'life1, 'async_trait>( &'life0 self, id: &'life1 InstrumentId, depth: u32, ) -> Pin<Box<dyn Future<Output = Result<OrderBook, Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Reads the order book once, to depth levels per side.

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

where 'life0: 'async_trait, 'life1: 'async_trait, MarketDataConnector: 'async_trait,

Starts a stream of the order book of one instrument.

fn unsubscribe<'life0, 'async_trait>( &'life0 self, sub: SubscriptionId, ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>

where 'life0: 'async_trait, MarketDataConnector: 'async_trait,

Stops a subscription of this connector.