Trait OrderBookSource 

pub trait OrderBookSource: Send + Sync {
    // Required methods
    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 Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    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 Self: 'async_trait,
             'life0: 'async_trait,
             'life1: 'async_trait;
    fn unsubscribe<'life0, 'async_trait>(
        &'life0 self,
        sub: SubscriptionId,
    ) -> Pin<Box<dyn Future<Output = Result<(), Error>> + Send + 'async_trait>>
       where Self: 'async_trait,
             'life0: 'async_trait;
}

Order book snapshot and stream.

order_book reads the book once and answers an OrderBook::Snapshot. subscribe_order_book starts a stream: the books arrive as Event::OrderBook in the channel given to the connector at start, tagged with the returned SubscriptionId. After a reconnect the stream sends Event::Resync and a new snapshot. unsubscribe stops a subscription of this connector by its id.

Examples

use std::time::Duration;
use brokapi_bcs::limits::default_limits;
use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
use brokapi_core::{
    Event, InstrumentId, InstrumentKey, NativeId, OrderBook, OrderBookSource, Price, ShutdownMode,
    SubscribeOpts,
};
use brokapi_runtime::BrokerLimiter;

/// Works with the connector of any broker.
async fn best_bid(src: &impl OrderBookSource, id: &InstrumentId) -> Option<Price> {
    match src.order_book(id, 5).await.ok()? {
        OrderBook::Snapshot { bids, .. } => bids.first().map(|l| l.price),
        _ => None,
    }
}

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() },
);
assert!(best_bid(&md, &sber).await.is_some());

// A stream: the book arrives as an envelope with the id of the subscription.
let sub = md.subscribe_order_book(&sber, 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));
assert!(matches!(env.payload, Event::OrderBook(_)));

md.unsubscribe(sub).await?;
md.shutdown(ShutdownMode::Immediate).await;

Required Methods

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

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

The answer is an OrderBook::Snapshot. Bids are listed best (highest) first and asks best (lowest) first. A depth above MarketDataCaps::max_order_book_depth is cut to it (the BCS connector does so without telling).

Errors

• Error::Config when id.native is of a kind this broker does not use.
• Error::RateLimited when the local limiter or the broker refuses for now. Wait and repeat.
• Error::Transport when the network failed or the answer was unreadable. Repeat later.
• Error::Rejected when the broker refuses the request, for example an unknown instrument (BCS: code validation).

Examples

use brokapi_core::{ErrorClass, OrderBook, OrderBookSource};
let OrderBook::Snapshot { bids, asks, .. } = md.order_book(&sber, 3).await? else {
    panic!("a read always answers a snapshot");
};
assert!(bids.len() <= 3 && asks.len() <= 3);
// The best prices are first on both sides, so the spread is the gap between them.
assert!(bids[0].price < asks[0].price);

// An instrument id of another broker's kind is a setup error: the call is not made.
let wrong = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Figi("BBG004730N88".into()));
let err = md.order_book(&wrong, 3).await.unwrap_err();
assert!(matches!(err, Error::Config(_)));
assert_eq!(err.class(), ErrorClass::Fatal);

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

Starts a stream of the order book of one instrument.

Returns when the stream is open at the broker (see Contract). Books then arrive as Event::OrderBook with Envelope::sub equal to the returned id. The stream restores itself after a reconnect and announces it with Event::Resync. opts.depth is the number of levels (None: the connector’s default, BCS uses its maximum). For books use Backpressure::Conflate: a slow reader then sees the latest book and not a queue of stale ones.

Contract

• When: at any time, no reconcile is needed (the market data connector has no trading gate).
• Guarantee: the call returns only when the stream is open: the broker has confirmed the subscription, or, where the protocol has no confirmation, the connection of the stream is up. What the broker sends after the return reaches you. A stream that is not open within the connector’s subscribe_ack_timeout (3 seconds by default) is Error::Transport, and nothing is left half open, so the call can be repeated. (Finam returns Ok after the wait and goes on opening the stream, see OrderExecutor::subscribe_order_state.)
• Guarantee: the stream restores itself after a reconnect. Each restore is announced with Event::Resync: drop the book you hold for that instrument and wait for the next snapshot.

Errors

• Error::Config when id.native is of a kind this broker does not use.
• Error::Unsupported when the connector has no order book stream (StreamingCaps::order_book is false).
• Error::RateLimited, Error::Transport when the subscription could not be made, also when the stream did not open in time. The subscription does not exist, ask again later.
• Error::Rejected with code subscription_refused when the broker refused it (an unknown instrument, no permission). Repeating the call unchanged will not help.

Examples

use std::time::Duration;
use brokapi_core::{Backpressure, Event, OrderBookSource, SubscribeOpts};
// The 10 best levels, latest book wins if we read slowly.
let opts = SubscribeOpts { depth: Some(10), ..SubscribeOpts::conflate() };
assert_eq!(opts.backpressure, Backpressure::Conflate);
let sub = md.subscribe_order_book(&sber, opts).await?;

// All subscriptions of the connector write into the one channel, so route by `sub`.
let env = tokio::time::timeout(Duration::from_secs(8), events.recv()).await?.expect("channel is open");
assert_eq!(env.sub, Some(sub));
let Event::OrderBook(book) = env.payload else { panic!("an order book subscription sends books") };

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

where Self: 'async_trait, 'life0: 'async_trait,

Stops a subscription of this connector.

After it returns, no new events of sub are produced. Envelopes already in your channel stay there, so keep reading until you have what you need. A subscription ends on its own when the connector shuts down.

Errors

• Error::Rejected with code unknown_subscription when sub is not an active subscription of this connector: it was never given out, or it is already stopped. The same at all brokers.
• Error::Transport when the broker could not be told.

Examples

use brokapi_core::{OrderBookSource, SubscribeOpts};
let sub = md.subscribe_order_book(&sber, SubscribeOpts::conflate()).await?;
md.unsubscribe(sub).await?;
// The id is gone: a second unsubscribe is an error, not a silent success.
assert!(md.unsubscribe(sub).await.is_err());

Dyn Compatibility

This trait is dyn compatible.

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

Implementors