brokapi_alor/
md.rs

1//! The market data connector: order books, trades of the market, quotes, candles and the reference data of ALOR.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use brokapi_alor_protocol::http::client::AlorHttpClient;
9use brokapi_alor_protocol::ws::models::WsSubscribe;
10use brokapi_alor_protocol::ws::session::WsSession;
11use brokapi_core::{
12    Candle, Envelope, Error, Instrument, InstrumentId, InstrumentKind, InstrumentQuery,
13    InstrumentSource, MarketDataCaps, MarketDataSource, MarketTrade, OrderBook, OrderBookSource,
14    Schedule, ShutdownMode, ShutdownReport, StreamingCaps, SubscribeOpts, SubscriptionId,
15    TimeRange, Timeframe, TradingParams, TradingStatus,
16};
17use brokapi_runtime::{BrokerLimiter, Emitter};
18use brokapi_sansio::SubscriptionKind;
19use tokio::sync::{Mutex, RwLock, mpsc};
20use tokio::task::JoinHandle;
21use tokio_util::sync::CancellationToken;
22
23use crate::config::MarketDataConfig;
24use crate::core::MdCore;
25use crate::error::{map_config_err, map_err};
26use crate::map::instrument::{instrument, trading_params, trading_status};
27use crate::map::market::{candles, market_trade, order_book};
28use crate::map::unix_nanos;
29use crate::pump::{drain_backlog, run_pump};
30use crate::raw::AlorRaw;
31use crate::subs::{MdSubKind, MdSubscription, SubIdGen, new_guid, wait_ack};
32use crate::util::{exchange_of_board, nanos_to_secs, ticker_board, timeframe_label};
33
34/// The most levels of an order book the broker gives (50 on each side).
35const MAX_DEPTH: u32 = 50;
36
37/// How many instruments a page of the search asks for, and how many pages are read at most.
38const KIND_PAGE_SIZE: u32 = 100;
39const MAX_KIND_PAGES: u32 = 200;
40
41struct MdShared {
42    http: AlorHttpClient,
43    ws: WsSession,
44    limiter: Arc<BrokerLimiter>,
45    core: Arc<Mutex<MdCore>>,
46    emitter: Arc<Emitter>,
47    subs: RwLock<HashMap<SubscriptionId, MdSubscription>>,
48    sub_ids: SubIdGen,
49    cancel: CancellationToken,
50    ack_timeout: Duration,
51}
52
53/// The market data connector of ALOR: implements [`OrderBookSource`], [`MarketDataSource`] and
54/// [`InstrumentSource`]. It is not tied to an account.
55///
56/// Start it with [`MarketDataConnector::start`]; commands return their answer directly and subscriptions deliver
57/// [`Envelope`]s into the channel given at start. The data comes over one WebSocket session that reconnects on its
58/// own and restores the subscriptions (an order book is then marked with `Event::Resync`).
59///
60/// An instrument is named by `NativeId::TickerClass { ticker, class_code }`, where the class code is the trading
61/// mode (the board, for example `TQBR`). The exchange follows from it: the board `SPBX` is the exchange `SPBX`,
62/// every other board is `MOEX`.
63///
64/// # Examples
65///
66/// ```
67/// # #[tokio::main(flavor = "current_thread")]
68/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
69/// # let server = brokapi_mocks::alor::MockServer::start(Default::default()).await;
70/// # let cfg = brokapi_alor_protocol::config::AlorConfig::for_mock("mock-refresh-token", &server.http_base);
71/// use brokapi_alor::limits::default_limits;
72/// use brokapi_alor::util::instrument_id;
73/// use brokapi_alor::{MarketDataConfig, MarketDataConnector};
74/// use brokapi_core::{OrderBook, OrderBookSource, ShutdownMode};
75/// use brokapi_runtime::BrokerLimiter;
76/// use tokio::sync::mpsc;
77///
78/// let (tx, _events) = mpsc::channel(256);
79/// let md = MarketDataConnector::start(
80///     MarketDataConfig::new(cfg, "alor-md"),
81///     BrokerLimiter::new(&default_limits()),
82///     tx,
83/// )
84/// .await?;
85/// let book = md.order_book(&instrument_id("SBER", "TQBR"), 5).await?;
86/// let OrderBook::Snapshot { bids, .. } = book else { unreachable!() };
87/// assert!(!bids.is_empty());
88/// md.shutdown(ShutdownMode::Immediate).await;
89/// # Ok(())
90/// # }
91/// ```
92pub struct MarketDataConnector {
93    inner: Arc<MdShared>,
94    pump: JoinHandle<()>,
95}
96
97impl MarketDataConnector {
98    /// Starts the connector: gets the first access token (so a refused token is an error here, not at the first
99    /// call) and opens the WebSocket session in the background.
100    ///
101    /// # Errors
102    ///
103    /// [`Error::Config`] when the client cannot be built or the broker refuses the token; [`Error::Transport`]
104    /// when the authorization server cannot be reached.
105    pub async fn start(
106        cfg: MarketDataConfig,
107        limiter: Arc<BrokerLimiter>,
108        out: mpsc::Sender<Envelope>,
109    ) -> Result<Self, Error> {
110        let http = AlorHttpClient::new(cfg.alor.clone()).map_err(map_config_err)?;
111        http.get_access_token().await.map_err(map_err)?;
112
113        let (event_tx, event_rx) = kanal::bounded_async(512);
114        let ws = WsSession::spawn_with(http.clone(), event_tx, cfg.ws_options());
115        let cancel = CancellationToken::new();
116        let core = Arc::new(Mutex::new(MdCore::new()));
117        let emitter = Emitter::new(Arc::clone(&cfg.source), out);
118
119        let inner = Arc::new(MdShared {
120            http,
121            ws,
122            limiter,
123            core: Arc::clone(&core),
124            emitter: Arc::clone(&emitter),
125            subs: RwLock::new(HashMap::new()),
126            sub_ids: SubIdGen::new(0),
127            cancel: cancel.clone(),
128            ack_timeout: cfg.subscribe_ack_timeout,
129        });
130        let pump = tokio::spawn(run_pump(
131            core,
132            emitter,
133            event_rx,
134            cancel,
135            |m| async move { m },
136        ));
137        Ok(Self { inner, pump })
138    }
139
140    /// Direct access to any API method with this connector's token and rate limiter. See [`AlorRaw`].
141    pub fn raw(&self) -> AlorRaw {
142        AlorRaw::new(self.inner.http.clone(), Arc::clone(&self.inner.limiter))
143    }
144
145    /// What the connector can do: books of up to 50 levels (always whole books), all four streams.
146    pub fn caps(&self) -> MarketDataCaps {
147        MarketDataCaps::new(MAX_DEPTH, false, StreamingCaps::new(true, true, true, true))
148    }
149
150    /// Stops the connector. Unless the mode is `Immediate`, it cancels the subscriptions at the broker first;
151    /// `Graceful` also lets the bot's channel take what is still buffered.
152    pub async fn shutdown(self, mode: ShutdownMode) -> ShutdownReport {
153        let mut unsubscribed = 0usize;
154        let mut unsubscribe_failed = 0usize;
155        if !matches!(mode, ShutdownMode::Immediate) {
156            let subs: Vec<_> = self.inner.subs.read().await.values().cloned().collect();
157            for sub in subs {
158                match self.inner.ws.unsubscribe(&sub.guid).await {
159                    Ok(()) => unsubscribed += 1,
160                    Err(_) => unsubscribe_failed += 1,
161                }
162            }
163        }
164        if let ShutdownMode::Graceful { deadline } = mode {
165            drain_backlog(&self.inner.emitter, deadline).await;
166        }
167        self.inner.cancel.cancel();
168        let _ = self.pump.await;
169        ShutdownReport::new(unsubscribed, unsubscribe_failed, Vec::new())
170    }
171
172    fn acquire(&self, method: &str) -> Result<(), Error> {
173        self.inner.limiter.acquire(method, brokapi_runtime::now())
174    }
175
176    async fn register_md_sub(
177        &self,
178        instrument: InstrumentId,
179        kind: MdSubKind,
180        core_kind: SubscriptionKind,
181        opts: SubscribeOpts,
182        guid: String,
183        ws: WsSubscribe,
184    ) -> Result<SubscriptionId, Error> {
185        let id = self.inner.sub_ids.next();
186        self.inner
187            .emitter
188            .register(id, Some(instrument.clone()), opts.backpressure);
189        // The core attributes incoming data to this subscription. It is told before the request goes out, so the
190        // confirmation cannot overtake it.
191        self.inner.core.lock().await.register(&guid, id, core_kind);
192        let notify = self.inner.core.lock().await.notifier();
193        if let Err(e) = self.inner.ws.subscribe(ws).await {
194            self.inner.core.lock().await.forget(&guid);
195            self.inner.emitter.unregister(id);
196            return Err(map_err(e));
197        }
198        // Returns when the broker has confirmed the subscription: what happens after that is delivered.
199        let confirmed = wait_ack(
200            &self.inner.core,
201            notify,
202            self.inner.ack_timeout,
203            |c: &MdCore| {
204                (
205                    c.actual_state(id).unwrap_or(crate::core::SubState::Pending),
206                    c.failure(id).map(str::to_string),
207                )
208            },
209        )
210        .await;
211        if let Err(e) = confirmed {
212            let _ = self.inner.ws.unsubscribe(&guid).await;
213            self.inner.core.lock().await.forget(&guid);
214            self.inner.emitter.unregister(id);
215            return Err(e);
216        }
217        self.inner.subs.write().await.insert(
218            id,
219            MdSubscription {
220                id,
221                instrument,
222                kind,
223                opts,
224                guid,
225            },
226        );
227        Ok(id)
228    }
229}
230
231#[async_trait]
232impl brokapi_core::Connector for MarketDataConnector {
233    type Caps = brokapi_core::MarketDataCaps;
234
235    fn caps(&self) -> brokapi_core::MarketDataCaps {
236        MarketDataConnector::caps(self)
237    }
238
239    async fn shutdown(self, mode: brokapi_core::ShutdownMode) -> brokapi_core::ShutdownReport {
240        MarketDataConnector::shutdown(self, mode).await
241    }
242}
243
244#[async_trait]
245impl OrderBookSource for MarketDataConnector {
246    async fn order_book(&self, id: &InstrumentId, depth: u32) -> Result<OrderBook, Error> {
247        self.acquire("GetOrderBook")?;
248        let (ticker, board) = ticker_board(id)?;
249        let wire = self
250            .inner
251            .http
252            .get_order_book(
253                exchange_of_board(&board),
254                &ticker,
255                Some(&board),
256                Some(depth.clamp(1, MAX_DEPTH)),
257            )
258            .await
259            .map_err(map_err)?;
260        Ok(order_book(&wire, id, brokapi_runtime::now()))
261    }
262
263    async fn subscribe_order_book(
264        &self,
265        id: &InstrumentId,
266        opts: SubscribeOpts,
267    ) -> Result<SubscriptionId, Error> {
268        let depth = opts.depth.unwrap_or(MAX_DEPTH).clamp(1, MAX_DEPTH);
269        let (ticker, board) = ticker_board(id)?;
270        let guid = new_guid();
271        let ws = WsSubscribe::order_book(
272            &guid,
273            exchange_of_board(&board),
274            &ticker,
275            Some(&board),
276            depth,
277        );
278        self.register_md_sub(
279            id.clone(),
280            MdSubKind::OrderBook,
281            SubscriptionKind::OrderBook {
282                instrument: id.clone(),
283                depth,
284            },
285            opts,
286            guid,
287            ws,
288        )
289        .await
290    }
291
292    async fn unsubscribe(&self, sub: SubscriptionId) -> Result<(), Error> {
293        let record = self
294            .inner
295            .subs
296            .write()
297            .await
298            .remove(&sub)
299            .ok_or_else(|| Error::Rejected {
300                code: "unknown_subscription".into(),
301                message: format!("subscription {} is not active", sub.value()),
302            })?;
303        self.inner.core.lock().await.forget(&record.guid);
304        self.inner.emitter.unregister(sub);
305        self.inner
306            .ws
307            .unsubscribe(&record.guid)
308            .await
309            .map_err(map_err)
310    }
311}
312
313#[async_trait]
314impl MarketDataSource for MarketDataConnector {
315    async fn last_trades(&self, id: &InstrumentId, limit: u32) -> Result<Vec<MarketTrade>, Error> {
316        self.acquire("GetLastTrades")?;
317        let (ticker, board) = ticker_board(id)?;
318        let query = [
319            ("take", limit.to_string()),
320            ("descending", "true".to_string()),
321        ];
322        let records = self
323            .inner
324            .http
325            .get_all_trades(exchange_of_board(&board), &ticker, Some(&board), &query)
326            .await
327            .map_err(map_err)?;
328        let now = brokapi_runtime::now();
329        // Newest first from the broker; the shared interface wants the oldest first.
330        let mut trades: Vec<MarketTrade> = records
331            .iter()
332            .rev()
333            .filter_map(|r| market_trade(r, id, now))
334            .collect();
335        if trades.len() > limit as usize {
336            trades.drain(..trades.len() - limit as usize);
337        }
338        Ok(trades)
339    }
340
341    async fn subscribe_market_trades(
342        &self,
343        id: &InstrumentId,
344        opts: SubscribeOpts,
345    ) -> Result<SubscriptionId, Error> {
346        let (ticker, board) = ticker_board(id)?;
347        let guid = new_guid();
348        let ws = WsSubscribe::all_trades(&guid, exchange_of_board(&board), &ticker, Some(&board));
349        self.register_md_sub(
350            id.clone(),
351            MdSubKind::MarketTrades,
352            SubscriptionKind::LastTrades {
353                instrument: id.clone(),
354            },
355            opts,
356            guid,
357            ws,
358        )
359        .await
360    }
361
362    async fn subscribe_quotes(
363        &self,
364        id: &InstrumentId,
365        opts: SubscribeOpts,
366    ) -> Result<SubscriptionId, Error> {
367        let (ticker, board) = ticker_board(id)?;
368        let guid = new_guid();
369        let ws = WsSubscribe::quotes(&guid, exchange_of_board(&board), &ticker, Some(&board));
370        self.register_md_sub(
371            id.clone(),
372            MdSubKind::Quotes,
373            SubscriptionKind::Quotes {
374                instrument: id.clone(),
375            },
376            opts,
377            guid,
378            ws,
379        )
380        .await
381    }
382
383    async fn candles(
384        &self,
385        id: &InstrumentId,
386        tf: Timeframe,
387        range: TimeRange,
388    ) -> Result<Vec<Candle>, Error> {
389        self.acquire("GetCandles")?;
390        let (ticker, board) = ticker_board(id)?;
391        let resp = self
392            .inner
393            .http
394            .get_history(
395                exchange_of_board(&board),
396                &ticker,
397                Some(&board),
398                &timeframe_label(tf),
399                nanos_to_secs(range.from),
400                nanos_to_secs(range.to),
401                None,
402            )
403            .await
404            .map_err(map_err)?;
405        Ok(candles(&resp, id, tf, brokapi_runtime::now()))
406    }
407
408    async fn subscribe_candles(
409        &self,
410        id: &InstrumentId,
411        tf: Timeframe,
412        opts: SubscribeOpts,
413    ) -> Result<SubscriptionId, Error> {
414        let (ticker, board) = ticker_board(id)?;
415        let guid = new_guid();
416        // Only the candle that is forming now and the ones after it: the history is read with `candles`.
417        let from = nanos_to_secs(unix_nanos(chrono::Utc::now()));
418        let ws = WsSubscribe::bars(
419            &guid,
420            exchange_of_board(&board),
421            &ticker,
422            Some(&board),
423            &timeframe_label(tf),
424            from,
425            true,
426        );
427        self.register_md_sub(
428            id.clone(),
429            MdSubKind::Candles,
430            SubscriptionKind::LastCandle {
431                instrument: id.clone(),
432                timeframe: tf,
433            },
434            opts,
435            guid,
436            ws,
437        )
438        .await
439    }
440}
441
442#[async_trait]
443impl InstrumentSource for MarketDataConnector {
444    async fn find(&self, q: InstrumentQuery) -> Result<Vec<Instrument>, Error> {
445        self.acquire("GetSecurities")?;
446        match q {
447            InstrumentQuery::ByTicker { ticker, exchange } => {
448                let mut query = vec![("query", ticker.clone()), ("limit", "100".to_string())];
449                if let Some(e) = exchange {
450                    query.push(("exchange", e));
451                }
452                // The search matches a part of the ticker: keep the exact ticker, in every board it trades in.
453                query.push(("includeNonBaseBoards", "true".to_string()));
454                let list = self
455                    .inner
456                    .http
457                    .get_securities(&query)
458                    .await
459                    .map_err(map_err)?;
460                Ok(list
461                    .iter()
462                    .filter(|s| s.symbol.eq_ignore_ascii_case(&ticker))
463                    .map(instrument)
464                    .collect())
465            }
466            InstrumentQuery::ByIsin(_) => Err(Error::Unsupported {
467                capability: "find_by_isin".into(),
468            }),
469            InstrumentQuery::ByKind(kind) => self.find_by_kind(kind).await,
470        }
471    }
472
473    async fn get(&self, id: &InstrumentId) -> Result<Instrument, Error> {
474        self.acquire("GetSecurity")?;
475        let (ticker, board) = ticker_board(id)?;
476        let info = self
477            .inner
478            .http
479            .get_security(exchange_of_board(&board), &ticker, Some(&board))
480            .await
481            .map_err(map_err)?;
482        Ok(instrument(&info))
483    }
484
485    async fn trading_params(&self, id: &InstrumentId) -> Result<TradingParams, Error> {
486        self.acquire("GetSecurity")?;
487        let (ticker, board) = ticker_board(id)?;
488        let info = self
489            .inner
490            .http
491            .get_security(exchange_of_board(&board), &ticker, Some(&board))
492            .await
493            .map_err(map_err)?;
494        Ok(trading_params(&info))
495    }
496
497    async fn trading_status(&self, id: &InstrumentId) -> Result<TradingStatus, Error> {
498        self.acquire("GetTradingStatus")?;
499        let (ticker, board) = ticker_board(id)?;
500        let info = self
501            .inner
502            .http
503            .get_security(exchange_of_board(&board), &ticker, Some(&board))
504            .await
505            .map_err(map_err)?;
506        let mut status = trading_status(&info, Some(unix_nanos(chrono::Utc::now())));
507        // The status is about the instrument asked for, in the board asked for.
508        status.instrument = id.clone();
509        Ok(status)
510    }
511
512    async fn schedule(&self, _exchange: &str, _range: TimeRange) -> Result<Schedule, Error> {
513        // ALOR gives the state of a market now (`/md/status/...`) and the status code of an instrument, but no
514        // timetable of sessions.
515        Err(Error::Unsupported {
516            capability: "schedule".into(),
517        })
518    }
519}
520
521impl MarketDataConnector {
522    /// Instruments of a kind: a search by market, filtered by the kind the connector reads from the type.
523    async fn find_by_kind(&self, kind: InstrumentKind) -> Result<Vec<Instrument>, Error> {
524        let sector = match kind {
525            InstrumentKind::Stock
526            | InstrumentKind::Bond
527            | InstrumentKind::Etf
528            | InstrumentKind::MutualFund => "FOND",
529            InstrumentKind::Future | InstrumentKind::Option => "FORTS",
530            InstrumentKind::Currency => "CURR",
531            InstrumentKind::Perpetual => {
532                return Err(Error::Unsupported {
533                    capability: "find_perpetuals_by_kind".into(),
534                });
535            }
536            InstrumentKind::Other => {
537                return Err(Error::Unsupported {
538                    capability: "find_other_by_kind".into(),
539                });
540            }
541        };
542        let mut all = Vec::new();
543        for page in 0..MAX_KIND_PAGES {
544            if page > 0 {
545                self.acquire("GetSecurities")?;
546            }
547            let query = [
548                ("sector", sector.to_string()),
549                ("limit", KIND_PAGE_SIZE.to_string()),
550                ("offset", (page * KIND_PAGE_SIZE).to_string()),
551            ];
552            let batch = self
553                .inner
554                .http
555                .get_securities(&query)
556                .await
557                .map_err(map_err)?;
558            let full = batch.len() >= KIND_PAGE_SIZE as usize;
559            all.extend(
560                batch
561                    .iter()
562                    .filter(|s| crate::map::instrument::kind_of(s) == kind)
563                    .map(instrument),
564            );
565            if !full {
566                return Ok(all);
567            }
568        }
569        Err(Error::Transport(brokapi_core::TransportError::Message(
570            format!(
571                "the list of instruments of this kind is longer than {MAX_KIND_PAGES} pages: not returned in part"
572            ),
573        )))
574    }
575}
brokapi_alor/md.rs

brokapi_alor/
md.rs
