Struct MarketDataConnector 

pub struct MarketDataConnector { /* private fields */ }

T-Bank market data connector: order books, trades, quotes, candles and the instrument directory.

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

How T-Bank differs from the other brokers:

• All market data streams run over ONE bidirectional gRPC stream, which the connector keeps open and reopens with a growing delay after a break. After a reopen the subscriptions are sent again and an order book subscription gets an Event::Resync.
• Every order book message is a full snapshot, there are no increments.
• Quotes are the best level of the order book. T-Bank has no separate quote stream.
• The broker counts the levels of the order book, trades and candle volumes in LOTS. The connector multiplies by the lot size, so the events count in units of the instrument, as at every broker (the first read or subscription on an instrument looks the lot size up, one GetInstrumentBy). Prices are per one instrument.
• Depths the broker accepts are 1, 10, 20, 30, 40 and 50. Other depths are rounded up on the wire and the events are cut to the depth you asked for.

Examples

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

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

// The uid comes from the instrument directory, see `InstrumentSource::find`.
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));

// A snapshot by request.
let OrderBook::Snapshot { bids, asks, .. } = md.order_book(&sber, 10).await? else {
    panic!("T-Bank answers a snapshot")
};
assert!(!bids.is_empty() && !asks.is_empty());

// A stream. The call returns once the broker has confirmed the subscription.
let sub = md.subscribe_order_book(&sber, SubscribeOpts::conflate()).await?;
let env = tokio::time::timeout(Duration::from_secs(5), async {
    loop {
        let env = events.recv().await.expect("the channel stays open");
        if matches!(env.payload, Event::OrderBook(_)) {
            break env;
        }
    }
})
.await?;
assert_eq!(env.sub, Some(sub));

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

Implementations

impl MarketDataConnector

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

Builds the channel and starts the stream task.

The task opens the market data stream in the background at once and reopens it after a break (delay from 500 ms doubling to 30 s). start does not wait for it and does not fail when the broker cannot be reached: the first request or subscription reports that. A subscription waits for the broker to confirm it, for at most request_timeout.

limiter should be the one limiter of the whole user, shared with every other connector (see crate::limits). Events go to out.

Errors

[Error::Config] when the token or app name is not a valid header value, or the endpoint is not a valid URL. Nothing was started in that case.

Examples

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

let limiter = BrokerLimiter::new(&default_limits());
let (events_tx, _events) = mpsc::channel(256);

// A bad endpoint is a configuration error at start.
let mut bad = cfg.clone();
bad.endpoint = "not a url".into();
let refused = MarketDataConnector::start(MarketDataConfig::new(bad, "tbank-md"), limiter.clone(), events_tx.clone()).await;
assert!(matches!(refused, Err(Error::Config(_))));

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

pub fn raw(&self) -> TbankRaw

Any other method of the T-Bank API with this connector’s token, endpoint and rate limiter. See TbankRaw.

The raw layer is not blocked by the connector’s shutdown gate and does not know the connector’s subscriptions. Its calls spend quota from the same limiter as the connector’s own.

Examples

use brokapi_runtime::BrokerLimiter;
use brokapi_tbank::limits::default_limits;
use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
use brokapi_tbank_protocol::v1 as pb;
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(64);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;

// Last prices of several instruments at once: no shared trait has this call.
let prices = md
    .raw()
    .call("GetLastPrices", |c| async move {
        c.market_data()
            .get_last_prices(pb::GetLastPricesRequest {
                instrument_id: vec!["SBER_TQBR".into()],
                ..Default::default()
            })
            .await
    })
    .await?;
assert_eq!(prices.last_prices.len(), 1);

pub fn caps(&self) -> MarketDataCaps

What this connector can do.

The order book is at most 50 levels deep ([MarketDataCaps::max_order_book_depth]). Every stream message is a full snapshot (incremental_order_book is false). Quotes are the best level of the book, since T-Bank has no separate quote stream. Candles stream as they form.

Examples

use brokapi_runtime::BrokerLimiter;
use brokapi_tbank::limits::default_limits;
use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(64);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let caps = md.caps();
assert_eq!(caps.max_order_book_depth, 50);
assert!(!caps.incremental_order_book);
assert!(caps.streaming.quotes && caps.streaming.candles);

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

Stops the connector and its stream task.

With [ShutdownMode::Graceful] the connector first unsubscribes every wire subscription and waits for the broker to confirm, until the deadline. Subscriptions not confirmed by then count as unsubscribe_failed. It also gives the event backlog the same deadline to drain into your channel. With [ShutdownMode::Immediate] it just stops, and the report counts nothing.

Market data has no commands, so inflight_abandoned is always empty. The method consumes the connector, so no further call is possible. Closing the stream is the only cleanup the broker needs.

Examples

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

let (events_tx, _events) = mpsc::channel(256);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));
md.subscribe_order_book(&sber, SubscribeOpts::conflate()).await?;

let report = md.shutdown(ShutdownMode::Graceful { deadline: Duration::from_secs(2) }).await;
assert!(report.inflight_abandoned.is_empty());
println!("unsubscribed {}, failed {}", report.unsubscribed, report.unsubscribe_failed);

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 Self: '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 find<'life0, 'async_trait>( &'life0 self, q: InstrumentQuery, ) -> Pin<Box<dyn Future<Output = Result<Vec<Instrument>, Error>> + Send + 'async_trait>>

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

Searches the instrument directory by ticker, ISIN or kind.

ByTicker / ByIsin use FindInstrument (short records: currency is empty, use get). ByKind calls the listing of that kind: Shares, Bonds, Etfs, Futures, Options, Currencies (base list, the instruments tradable through the API). They are heavy (a few thousand records) and limited to 15 requests a minute: the limiter has its own bucket instruments_list, nothing is cached here. MutualFund / Other are Unsupported: T-Bank has no listing for them.

Errors

• [Error::Unsupported] for MutualFund and Other, without a network call.
• [Error::RateLimited] when a quota is used up. The listing quota is small (15 a minute).
• [Error::Transport], [Error::Config] and [Error::Rejected] as for order_book.

Examples

use brokapi_core::{
    Error, InstrumentId, InstrumentKey, InstrumentKind, InstrumentQuery, InstrumentSource, NativeId,
    ShutdownMode,
};
use brokapi_runtime::BrokerLimiter;
use brokapi_tbank::limits::default_limits;
use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));

// Search by ticker. A short record has no currency: use `get` for the full one.
let found = md
    .find(InstrumentQuery::ByTicker { ticker: "SBER".into(), exchange: None })
    .await?;
assert_eq!(found[0].id.key.symbol, "SBER");
let full = md.get(&found[0].id).await?;
assert_eq!(full.currency, "rub");

let params = md.trading_params(&sber).await?;
assert_eq!(params.min_qty.to_string(), "10");

// T-Bank has no listing of mutual funds.
let funds = md.find(InstrumentQuery::ByKind(InstrumentKind::MutualFund)).await;
assert!(matches!(funds, Err(Error::Unsupported { .. })));

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

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

The full record of one instrument (GetInstrumentBy), including its currency.

Errors

[Error::Rejected] (code not_found) when the broker does not know the instrument, and the other errors of order_book.

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

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

Lot size, price step and currency of an instrument.

lot_size is the units of the instrument in one lot. min_qty is one lot too: an order is a whole number of lots. price_step and price_step_value are both the instrument’s minimal price increment (for futures the step in money would need another call, which is not made).

Errors

The same as get.

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

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

Whether trading is going on now (GetTradingStatus).

The broker’s status is mapped to open, break, closed or unknown. Opening and closing periods and auctions are Unknown, because orders are possible there only with restrictions. The raw broker status name is in reason.

Errors

The same as get.

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

Trading sessions of an exchange in range (TradingSchedules).

exchange is the broker’s exchange name (for example MOEX). If the broker has no schedule with that name, the first one in the answer is used. An empty answer gives a schedule with no sessions.

Errors

The same as order_book.

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

Anonymous market trades of the last hour, the newest limit of them (GetLastTrades).

One hour is all that T-Bank guarantees, so older trades are not available here. Quantities are in units of the instrument (the connector converts from lots). Trades without a direction are skipped, so you may get fewer than limit.

Errors

The same as order_book.

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,

Streams anonymous market trades as they happen.

Each event is one trade. Quantities are in units of the instrument (the connector converts from lots). Trades without a direction are skipped.

Errors

The same as subscribe_order_book.

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,

Best bid and ask: the first level of the order book stream.

T-Bank has no quote stream of its own, so this subscribes to a one-level order book. Every order book message produces one quote event.

Errors

The same as subscribe_order_book.

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,

Historic candles of range (GetCandles).

close_ts of each candle is its open time plus the length of the timeframe (a month is taken as 30 days). Volumes are in units of the instrument (the connector converts from lots). How many candles one call returns, and the maximum range per timeframe, are broker limits that this method does not split: not verified against the live API.

Errors

The same as order_book.

Examples

use brokapi_core::{
    InstrumentId, InstrumentKey, MarketDataSource, NativeId, ShutdownMode, TimeRange, Timeframe, UnixNanos,
};
use brokapi_runtime::BrokerLimiter;
use brokapi_tbank::limits::default_limits;
use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));

let range = TimeRange::new(UnixNanos::from_nanos(1_700_000_000_000_000_000), UnixNanos::from_nanos(1_700_003_600_000_000_000));
let candles = md.candles(&sber, Timeframe::M5, range).await?;
assert!(candles.iter().all(|c| c.high >= c.low));

// Newest trades of the last hour, at most `limit` of them.
let trades = md.last_trades(&sber, 2).await?;
assert!(trades.len() <= 2);

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

The longest span one GetCandles request may cover, per timeframe, from the T-Invest documentation page “Исторические котировки” (1 minute: 1 day, 5 and 10 minutes: 1 week, 15 and 30 minutes: 3 weeks, 1 and 4 hours: 3 months, 1 day: 6 years, 1 week: 5 years, 1 month: 10 years). The broker calls the limits approximate. Not checked against the live API.

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,

Streams the candle that is forming, for timeframe tf.

Each event is the current state of the last candle, not only closed ones. Volumes are in units of the instrument (the connector converts from lots).

Errors

The same as subscribe_order_book.

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

One snapshot by request (GetOrderBook).

depth is the number of levels you want. It is sent rounded up to a depth the broker accepts (1, 10, 20, 30, 40 or 50) and the answer is cut back to depth levels, best first. A depth of 0 counts as 1 and a depth above 50 is capped at 50 on the wire. Quantities are in units of the instrument (the connector converts from lots).

Errors

• [Error::ShuttingDown] after shutdown started.
• [Error::RateLimited] when the local limiter or the broker refuses for rate. Repeat later.
• [Error::Rejected] when the broker does not know the instrument (code not_found).
• [Error::Transport] on a timeout or a network failure. A read, so repeat it later.
• [Error::Config] when the token is not accepted.

Examples

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

let (events_tx, _events) = mpsc::channel(256);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));

let OrderBook::Snapshot { bids, .. } = md.order_book(&sber, 3).await? else { panic!("a snapshot") };
assert!(bids.len() <= 3);

// An instrument the broker does not know is a refusal, not a transport error.
let nobody = InstrumentId::new(InstrumentKey::new("NOPE", "TQBR"), NativeId::Uid("no-such-uid".into()));
assert!(matches!(md.order_book(&nobody, 3).await, Err(Error::Rejected { .. })));

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,

Streams the order book as full snapshots.

opts.depth of None means 20 levels here (not the 50 maximum). Other values are clamped to 1..=50 and sent rounded up to a depth the broker accepts. Events are cut to the depth you asked for. Two subscriptions to one book share one wire subscription (the deeper one wins). The call returns after the broker confirmed the subscription. After a reconnect you get an Event::Resync and then fresh snapshots.

Errors

• [Error::ShuttingDown] after shutdown started.
• [Error::RateLimited] when the subscription quota (SubscribeMarketData) is used up.
• [Error::Rejected] when the broker refuses the subscription (the code is the broker’s subscription status, for example SUBSCRIPTION_STATUS_INSTRUMENT_NOT_FOUND).
• [Error::Transport] when the broker did not confirm within request_timeout (the message says why the stream is down, if it is).

Examples

use brokapi_core::{Error, InstrumentId, InstrumentKey, NativeId, OrderBookSource, ShutdownMode, SubscribeOpts};
use brokapi_runtime::BrokerLimiter;
use brokapi_tbank::limits::default_limits;
use brokapi_tbank::{MarketDataConfig, MarketDataConnector};
use tokio::sync::mpsc;

let (events_tx, _events) = mpsc::channel(256);
let md = MarketDataConnector::start(
    MarketDataConfig::new(cfg, "tbank-md"),
    BrokerLimiter::new(&default_limits()),
    events_tx,
)
.await?;
let sber = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Uid("uid-sber".into()));

let opts = SubscribeOpts { depth: Some(10), ..SubscribeOpts::conflate() };
let sub = md.subscribe_order_book(&sber, opts).await?;
md.unsubscribe(sub).await?;
// A second unsubscribe is a refusal: the subscription is gone.
assert!(matches!(md.unsubscribe(sub).await, Err(Error::Rejected { .. })));

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,

Ends a subscription made by any subscribe_* method.

The logical subscription is dropped at once and its buffer is closed. The wire subscription is dropped when no other subscription uses it. Events already queued for you may still arrive.

Errors

[Error::Rejected] with code unknown_subscription when the id is not active (never made, or already ended).