Struct MarketDataConnector 

pub struct MarketDataConnector { /* private fields */ }

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

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

What is specific to Finam:

• Instruments are SYMBOL@MIC. Pass NativeId::SymbolMic. Any other native id gives Error::Config without a network call.
• Order books come as full snapshots. Finam sends rows with ADD, UPDATE and REMOVE actions. The core folds them into a local book, so you never see increments (incremental_order_book is false). If an increment arrives with no snapshot under it, the stream is reopened and you get an Event::Resync.
• Depth. Finam documents no depth limit. The book is cut to the depth you ask for, at most caps().max_order_book_depth.
• Every call spends quota. Opening a stream counts as a request too. A refusal is Error::RateLimited, and nothing is sent.
• Streams reconnect by themselves with a growing pause (0.5 s up to 30 s). After a reconnect of an order book stream you get Event::Resync and then a new snapshot. Trades, quotes and candle streams resume without a notice (trades already seen are not repeated). sub_status says what a stream is doing.
• M10 candles are built by the connector from M5 candles, because the wire has no M10.
• Schedules are per instrument: pass SBER@MISX as the exchange argument of schedule.

Examples

Order book by request, then as a stream, with the instrument directory and an error path:

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

// `cfg` is your `FinamApiConfig`: the secret, and the endpoint of the real broker by default.
let (events_tx, mut events) = mpsc::channel(256);
let limiter = BrokerLimiter::new(&default_limits());
let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "finam-md"), limiter, events_tx).await?;

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "MISX"),
    NativeId::SymbolMic { symbol: "SBER".into(), mic: "MISX".into() },
);

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

// As a stream: envelopes arrive in `events`.
let sub = md.subscribe_order_book(&sber, SubscribeOpts::conflate()).await?;
let envelope = tokio::time::timeout(Duration::from_secs(5), events.recv()).await?.expect("channel open");
assert_eq!(envelope.sub, Some(sub));
assert!(matches!(envelope.payload, Event::OrderBook(_) | Event::Resync { .. }));
md.unsubscribe(sub).await?;

// The directory: one `GetAsset` when the exchange is known.
let found = md
    .find(InstrumentQuery::ByTicker { ticker: "SBER".into(), exchange: Some("MISX".into()) })
    .await?;
assert_eq!(found[0].id.key.symbol, "SBER");

// An instrument that is not a Finam symbol fails before anything is sent.
let foreign = InstrumentId::new(InstrumentKey::new("SBER", "TQBR"), NativeId::Figi("BBG004730N88".into()));
assert!(matches!(md.order_book(&foreign, 5).await, Err(Error::Config(_))));

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

Implementations

impl MarketDataConnector

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

Logs in, opens a lazy channel and starts the event pump. Streams are opened later, by subscribe_*.

limiter is the rate limiter of your Finam user: create one per user and share it with every connector. Events and subscription results go to out. Give it room, because a full channel makes a Lossless subscription overflow and a Conflate one merge events.

If cfg.api.tokens is set, that token manager is used and no login happens here.

Errors

• [Error::Config]: empty secret, or an invalid endpoint URL.
• [Error::Config] also: the broker refused the secret (a wrong token is a configuration problem at every broker).
• [Error::RateLimited]: the broker answered RESOURCE_EXHAUSTED to the login.
• [Error::Transport]: the broker could not be reached.

Examples

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

let limiter = BrokerLimiter::new(&default_limits());
let (events, _rx) = mpsc::channel(256);
let md = MarketDataConnector::start(MarketDataConfig::new(cfg.clone(), "finam-md"), limiter.clone(), events.clone()).await?;
md.shutdown(ShutdownMode::Immediate).await;

// A wrong secret is refused at start.
let wrong = FinamApiConfig { secret: "wrong".into(), ..cfg };
let err = MarketDataConnector::start(MarketDataConfig::new(wrong, "finam-md"), limiter, events).await.err().unwrap();
assert!(matches!(err, Error::Config(_)));

pub fn raw(&self) -> FinamRaw

Any other method of the Finam API with this connector’s token, endpoint and rate limiter. See FinamRaw.

Examples

See the raw module for a complete call.

pub fn caps(&self) -> MarketDataCaps

What this connector can do. Order books are streamed as full snapshots: Finam sends rows with ADD/UPDATE/REMOVE actions, which the core folds into a local book, so the consumer never sees increments (and Conflate is safe).

All four streams (order book, market trades, quotes, candles) are on. The maximum depth is the cut this connector applies, not a limit Finam documents.

Examples

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

let caps = md.caps();
assert!(caps.streaming.order_book && caps.streaming.candles);
assert!(!caps.incremental_order_book);

pub fn sub_status(&self, sub: SubscriptionId) -> Option<SubStatus>

State of a subscription’s stream (connecting, active, reconnecting, failed).

None for an id this connector does not know (never issued, or already unsubscribed). Use it after subscribe_* returned and the stream may still be opening: Finam may send the headers of a quiet stream only with its first message, so the call can return before the stream is open.

Examples

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

let sub = md.subscribe_order_book(&sber, SubscribeOpts::conflate()).await?;
// Wait for the first event: by then the stream is open.
tokio::time::timeout(Duration::from_secs(5), rx.recv()).await?;
assert_eq!(md.sub_status(sub), Some(SubStatus::Active));

md.unsubscribe(sub).await?;
assert_eq!(md.sub_status(sub), None);

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

Graceful: every stream is cancelled (the protocol-level unsubscribe of gRPC) and awaited within deadline; Immediate aborts them.

Consumes the connector and stops the event pump. The pump makes one last pass and hands over what fits into your channel without waiting. Events that do not fit are dropped. There are no commands here, so nothing is abandoned in flight (inflight_abandoned is always empty).

Examples

use std::time::Duration;
use brokapi_core::ShutdownMode;
use brokapi_finam::limits::default_limits;
use brokapi_finam::{FinamApiConfig, MarketDataConfig, MarketDataConnector};
use brokapi_runtime::BrokerLimiter;
use tokio::sync::mpsc;

let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "finam-md"), BrokerLimiter::new(&default_limits()), events).await?;
let report = md.shutdown(ShutdownMode::Graceful { deadline: Duration::from_secs(2) }).await;
assert_eq!(report.unsubscribe_failed, 0);

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,

ByTicker with an exchange is one GetAsset; everything else scans the (paged) directory of active instruments, so prefer the exchange form. The library keeps no cache.

The directory list carries no currency: currency is empty for the answers of a scan. Use get for it. ByKind matches the Finam type string by substring, which is not verified against the live API.

Errors

[Error::Rejected] (a symbol the broker does not know is not_found), [Error::RateLimited], [Error::Transport] (also when the directory has more pages than the safety bound of 200).

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,

One GetAsset: name, ISIN, kind and currency of the instrument.

Errors

[Error::Config] for a native id that is not SymbolMic. [Error::Rejected] when the broker does not know the symbol. [Error::RateLimited] and [Error::Transport] as for any read.

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,

Trading parameters from GetAsset (price step, lot size and the like). The library does not round your orders with them. Errors as in get.

The lot size comes from GetAsset (lot_size). When the token lists an account, a second call GetAssetParams for the first account asks for trade_lot_size, the lot size for trading. If it is above 0, it replaces lot_size and min_qty. If it is 0, missing, or the broker refuses that call, the lot of GetAsset stays. Not verified live: it is not known when the two differ.

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,

Derived from the instrument’s schedule: Finam has no direct status call.

Inside a session that is not of the closed type the status is open, otherwise closed. An empty schedule gives Unknown. Errors as in 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,

Finam schedules are per instrument, so exchange must be a symbol (SBER@MISX).

Only sessions that overlap range are returned, closed sessions are left out.

Errors

[Error::Config] when exchange has no @ (nothing is sent). Otherwise as in get.

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,

One LatestTrades request. Returns the newest limit trades, oldest first.

A trade that carries no side is dropped, because the core Side is mandatory.

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,

Opens a SubscribeLatestTrades stream. Trades seen before are not repeated after a reconnect (the last 2048 ids are remembered). Errors as in 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,

Opens a SubscribeQuote stream for one instrument. A side with price 0 is reported as missing. An in-band error message from the broker shows as SubStatus::Erroring. Errors as in 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,

One Bars request.

M10 is not on the wire: the connector asks for M5 and merges pairs. The start of the range moves down to the 10 minute boundary so the first candle keeps both halves. A lone M5 candle (one half had no trades) is a legitimate M10 candle and is kept.

Errors

As order_book, plus [Error::Unsupported] for a timeframe Finam has no bars for.

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

500 bars of the timeframe. The Finam documentation of Bars states no limit, so this is a cautious guess that costs only extra requests when it is too small. 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,

Opens a SubscribeBars stream. Every message of a forming bar gives an updated candle, not only the closed one. For M10 the stream is M5 and the connector builds the 10 minute candle. Errors as in candles and 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 OrderBook request. Answers OrderBook::Snapshot, cut to depth levels per side (at most 50, the cut of this connector).

Errors

• [Error::Config]: the instrument has no NativeId::SymbolMic. Nothing is sent.
• [Error::RateLimited]: the local quota is spent, or the broker answered RESOURCE_EXHAUSTED.
• [Error::Rejected]: the broker refused (unknown symbol, no permission).
• [Error::Transport]: connection failure or no answer within the request timeout.

See the type docs for an example.

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,

Opens an SubscribeOrderBook stream. Envelopes carry Event::OrderBook(OrderBook::Snapshot).

opts.depth is the number of levels per side (default and maximum 50). Waits up to subscribe_ack_timeout for the broker to open the stream: a refusal inside that time is returned. If the stream is not open by then the call still returns, and the stream goes on opening (see MarketDataConnector::sub_status): until it is open, nothing is delivered.

Errors

• [Error::Config]: the instrument has no NativeId::SymbolMic.
• [Error::Rejected]: the broker refused the stream (bad symbol, no permission), or this connector already has the maximum number of streams (code subscription_quota).

See the type docs for an example.

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,

Cancels the stream (the gRPC unsubscribe) and forgets the subscription. Waits up to 2 seconds for the stream task to end.

Errors

[Error::Rejected] with code unknown_subscription when sub is not a live subscription of this connector (a second unsubscribe of the same id ends here).