brokapi_bcs/md.rs
1//! The market data connector: order books, trades, quotes, candles and the instrument directory.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use brokapi_bcs_protocol::http::client::BcsHttpClient;
9use brokapi_bcs_protocol::http::models::{
10 CandlesRequest, DailyScheduleRequest, InstrumentsByIsinsRequest, InstrumentsByTickersRequest,
11 InstrumentsByTypeRequest, LastTradesRequest, OrderBookRequest, TradingStatusRequest,
12};
13use brokapi_bcs_protocol::ws::client::BcsWsClient;
14use brokapi_bcs_protocol::ws::models::MarketDataSubscribe;
15use brokapi_bcs_protocol::ws::session::WsSession;
16use brokapi_core::{
17 Candle, Envelope, Error, Instrument, InstrumentId, InstrumentQuery, InstrumentSource,
18 MarketDataCaps, MarketDataSource, MarketTrade, OrderBook, OrderBookSource, Schedule,
19 ShutdownMode, ShutdownReport, StreamingCaps, SubscribeOpts, SubscriptionId, TimeRange,
20 Timeframe, TradingParams, TradingStatus,
21};
22use brokapi_runtime::{AckState, BrokerLimiter, Emitter, wait_ack};
23use brokapi_sansio::{BrokerCore, DesiredSubscription, SubscriptionKind, SubscriptionSet};
24use tokio::sync::{Mutex, RwLock, mpsc};
25use tokio::task::JoinHandle;
26use tokio_util::sync::CancellationToken;
27
28use crate::config::MarketDataConfig;
29use crate::core::MdCore;
30use crate::error::{map_config_err, map_err};
31use crate::map::instrument::{map_instrument_info, map_trading_params, map_trading_status};
32use crate::map::market::{map_rest_candles, map_rest_last_trade, map_rest_order_book};
33use crate::map::schedule::map_daily_schedule;
34use crate::pump::{drain_backlog, run_md_pump};
35use crate::raw::BcsRaw;
36use crate::subs::{MdSubKind, MdSubscription, SubIdGen};
37use crate::util::{class_code, nanos_to_datetime, require_ticker_class, ticker, timeframe_label};
38
39const MAX_DEPTH: u32 = 20;
40
41/// `size` of `GET /instruments/by-type` (the docs allow 1..100, default 50).
42const KIND_PAGE_SIZE: u32 = 100;
43/// Pages read for one kind query at most. More is an error, never a silent cut.
44const MAX_KIND_PAGES: u32 = 200;
45
46struct MdShared {
47 http: BcsHttpClient,
48 ws: WsSession,
49 limiter: Arc<BrokerLimiter>,
50 core: Arc<Mutex<MdCore>>,
51 emitter: Arc<Emitter>,
52 subs: RwLock<HashMap<SubscriptionId, MdSubscription>>,
53 sub_ids: SubIdGen,
54 cancel: CancellationToken,
55 /// How long a `subscribe_*` call waits for the broker to confirm.
56 ack_timeout: Duration,
57}
58
59/// BCS market data connector: order books, trades, quotes, candles and the instrument directory.
60///
61/// One connector per broker is enough: it is not tied to an account. It implements
62/// [`OrderBookSource`], [`MarketDataSource`] and [`InstrumentSource`]. Commands return their
63/// answer directly. Subscriptions deliver [`Envelope`]s into the channel given to
64/// [`start`](Self::start).
65///
66/// # Examples
67///
68/// ```
69/// use brokapi_bcs::limits::default_limits;
70/// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
71/// use brokapi_core::{InstrumentId, InstrumentKey, NativeId, OrderBook, OrderBookSource};
72/// use brokapi_runtime::BrokerLimiter;
73/// use tokio::sync::mpsc;
74/// # use brokapi_bcs_protocol::config::BcsConfig;
75///
76/// # #[tokio::main(flavor = "current_thread")]
77/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
78/// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
79/// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
80/// let (events_tx, _events) = mpsc::channel(256);
81/// let limiter = BrokerLimiter::new(&default_limits());
82/// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
83///
84/// let sber = InstrumentId::new(
85/// InstrumentKey::new("SBER", "TQBR"),
86/// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
87/// );
88/// let book = md.order_book(&sber, 10).await?;
89/// let OrderBook::Snapshot { bids, asks, .. } = book else { panic!("REST answers a snapshot") };
90/// assert!(!bids.is_empty() && !asks.is_empty());
91/// # Ok(())
92/// # }
93/// ```
94pub struct MarketDataConnector {
95 inner: Arc<MdShared>,
96 pump: JoinHandle<()>,
97}
98
99impl MarketDataConnector {
100 /// Starts the connector: gets the first access token and opens the market data WebSocket.
101 ///
102 /// The refresh token in `cfg` is exchanged for an access token before anything else, so a wrong
103 /// token fails here and not on the first request. This step does not spend rate limit quota.
104 /// The WebSocket session is started in the background. It reconnects on its own and restores the
105 /// subscriptions after a reconnect. A reconnect voids the order books: the stream tells the bot
106 /// with `Event::Resync` before the new snapshot.
107 ///
108 /// `limiter` is shared by value of `Arc`: give the same limiter to every connector of one BCS
109 /// account, because the limits belong to the token. `out` receives the envelopes of all
110 /// subscriptions. Its capacity is the bot's choice. The bot must read it, because a slow reader
111 /// triggers the backpressure policy of each subscription.
112 ///
113 /// # Errors
114 ///
115 /// - [`Error::Config`] when the HTTP or WebSocket client cannot be built from `cfg`.
116 /// - [`Error::Transport`] when the token refresh fails: no connection, a timeout, or the auth
117 /// server refuses the refresh token. A refused or expired refresh token is a permanent failure,
118 /// it will not heal on retry (exact auth server answers: not verified against the live API).
119 /// - [`Error::RateLimited`] when the auth server answers HTTP 429.
120 ///
121 /// # Examples
122 ///
123 /// ```
124 /// use brokapi_bcs::limits::default_limits;
125 /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
126 /// use brokapi_core::ShutdownMode;
127 /// use brokapi_runtime::BrokerLimiter;
128 /// use tokio::sync::mpsc;
129 /// # use brokapi_bcs_protocol::config::BcsConfig;
130 ///
131 /// # #[tokio::main(flavor = "current_thread")]
132 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
133 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
134 /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
135 /// let (events_tx, _events) = mpsc::channel(256);
136 /// let limiter = BrokerLimiter::new(&default_limits());
137 /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
138 /// assert_eq!(md.caps().max_order_book_depth, 20);
139 /// md.shutdown(ShutdownMode::Immediate).await;
140 /// # Ok(())
141 /// # }
142 /// ```
143 ///
144 /// A broker that cannot be reached is a [`Error::Transport`], and nothing is left running:
145 ///
146 /// ```
147 /// use brokapi_bcs::limits::default_limits;
148 /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
149 /// use brokapi_core::Error;
150 /// use brokapi_runtime::BrokerLimiter;
151 /// use tokio::sync::mpsc;
152 /// use brokapi_bcs_protocol::config::BcsConfig;
153 ///
154 /// # #[tokio::main(flavor = "current_thread")]
155 /// # async fn main() {
156 /// // Port 1 of the local host: nothing listens there.
157 /// let cfg = BcsConfig::for_mock("any-token", "http://127.0.0.1:1");
158 /// let (events_tx, _events) = mpsc::channel(16);
159 /// let limiter = BrokerLimiter::new(&default_limits());
160 /// let started = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await;
161 /// assert!(matches!(started, Err(Error::Transport(_))));
162 /// # }
163 /// ```
164 pub async fn start(
165 cfg: MarketDataConfig,
166 limiter: Arc<BrokerLimiter>,
167 out: mpsc::Sender<Envelope>,
168 ) -> Result<Self, Error> {
169 let ws_client = BcsWsClient::new(cfg.bcs.clone())
170 .map_err(map_config_err)?
171 .with_options(cfg.ws_options());
172 ws_client
173 .http_client()
174 .authenticate()
175 .await
176 .map_err(map_err)?;
177
178 let (event_tx, event_rx) = kanal::bounded_async(512);
179 let ws = ws_client.connect_market_data(event_tx);
180 let cancel = CancellationToken::new();
181 let core = Arc::new(Mutex::new(MdCore::new()));
182 let emitter = Emitter::new(Arc::clone(&cfg.source), out);
183
184 let inner = Arc::new(MdShared {
185 http: ws_client.http_client().clone(),
186 ws,
187 limiter,
188 core: Arc::clone(&core),
189 emitter: Arc::clone(&emitter),
190 subs: RwLock::new(HashMap::new()),
191 sub_ids: SubIdGen::new(0),
192 cancel: cancel.clone(),
193 ack_timeout: cfg.subscribe_ack_timeout,
194 });
195
196 let pump = tokio::spawn(run_md_pump(core, emitter, event_rx, cancel));
197
198 Ok(Self { inner, pump })
199 }
200
201 /// Returns a handle for any other method of the BCS API, with this connector's token, endpoint and
202 /// rate limiter.
203 ///
204 /// The handle is cheap to clone and stays usable while the connector lives. Every call through it
205 /// spends quota by the operation name you give. See [`BcsRaw`] for the rules of a call.
206 ///
207 /// # Examples
208 ///
209 /// A method the shared traits do not wrap, here the quotes snapshot:
210 ///
211 /// ```
212 /// use brokapi_bcs_protocol::http::models::{QuoteInstrument, QuotesRequest};
213 /// use brokapi_bcs::limits::default_limits;
214 /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
215 /// use brokapi_core::ShutdownMode;
216 /// use brokapi_runtime::BrokerLimiter;
217 /// use tokio::sync::mpsc;
218 /// # use brokapi_bcs_protocol::config::BcsConfig;
219 ///
220 /// # #[tokio::main(flavor = "current_thread")]
221 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
222 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
223 /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
224 /// # let (events_tx, _events) = mpsc::channel(256);
225 /// # let limiter = BrokerLimiter::new(&default_limits());
226 /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
227 ///
228 /// let quotes = md
229 /// .raw()
230 /// .call("GetQuotes", |http| async move {
231 /// http.get_quotes(&QuotesRequest { instruments: vec![QuoteInstrument::new("SBER", "TQBR")] }).await
232 /// })
233 /// .await?;
234 /// assert!(!quotes.records.is_empty());
235 /// md.shutdown(ShutdownMode::Immediate).await;
236 /// # Ok(())
237 /// # }
238 /// ```
239 pub fn raw(&self) -> BcsRaw {
240 BcsRaw::new(self.inner.http.clone(), Arc::clone(&self.inner.limiter))
241 }
242
243 /// Describes what this connector can do, without any request.
244 ///
245 /// The answer is fixed for BCS: order books up to 20 levels, always as snapshots
246 /// (`incremental_order_book` is `false`), and streams for order books, market trades, quotes and
247 /// candles. A deeper `depth` in a request is cut to the maximum, it is not an error.
248 ///
249 /// # Examples
250 ///
251 /// ```
252 /// use brokapi_bcs::limits::default_limits;
253 /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
254 /// use brokapi_core::ShutdownMode;
255 /// use brokapi_runtime::BrokerLimiter;
256 /// use tokio::sync::mpsc;
257 /// # use brokapi_bcs_protocol::config::BcsConfig;
258 ///
259 /// # #[tokio::main(flavor = "current_thread")]
260 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
261 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
262 /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
263 /// let (events_tx, _events) = mpsc::channel(256);
264 /// let limiter = BrokerLimiter::new(&default_limits());
265 /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
266 /// let caps = md.caps();
267 /// assert_eq!(caps.max_order_book_depth, 20);
268 /// assert!(!caps.incremental_order_book);
269 /// assert!(caps.streaming.order_book && caps.streaming.candles);
270 /// # md.shutdown(ShutdownMode::Immediate).await;
271 /// # Ok(())
272 /// # }
273 /// ```
274 pub fn caps(&self) -> MarketDataCaps {
275 MarketDataCaps::new(MAX_DEPTH, false, StreamingCaps::new(true, true, true, true))
276 }
277
278 /// Stops the connector and reports what was cleaned up. Consumes the connector.
279 ///
280 /// - [`ShutdownMode::Immediate`]: no unsubscribe messages are sent, the WebSocket just goes away.
281 /// - [`ShutdownMode::Graceful`]: every subscription is unsubscribed on the WebSocket first. The
282 /// counts are in the report (`unsubscribed`, `unsubscribe_failed`). Then the events that still
283 /// wait in the subscription buffers get time to reach the bot's channel, until `deadline`.
284 ///
285 /// After the call the event pump has stopped, and no more envelopes arrive. Market data holds no
286 /// orders, so `inflight_abandoned` is always empty here.
287 ///
288 /// # Examples
289 ///
290 /// A graceful shutdown with one live subscription:
291 ///
292 /// ```
293 /// use std::time::Duration;
294 /// use brokapi_bcs::limits::default_limits;
295 /// use brokapi_bcs::{MarketDataConfig, MarketDataConnector};
296 /// use brokapi_core::{Backpressure, InstrumentId, InstrumentKey, NativeId, OrderBookSource, ShutdownMode, SubscribeOpts};
297 /// use brokapi_runtime::BrokerLimiter;
298 /// use tokio::sync::mpsc;
299 /// # use brokapi_bcs_protocol::config::BcsConfig;
300 ///
301 /// # #[tokio::main(flavor = "current_thread")]
302 /// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
303 /// # let server = brokapi_mocks::bcs::MockServer::start(Default::default()).await;
304 /// # let cfg = BcsConfig::for_mock("mock-refresh-token", &server.http_base);
305 /// # let (events_tx, _events) = mpsc::channel(256);
306 /// # let limiter = BrokerLimiter::new(&default_limits());
307 /// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bcs-md"), limiter, events_tx).await?;
308 /// let sber = InstrumentId::new(
309 /// InstrumentKey::new("SBER", "TQBR"),
310 /// NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
311 /// );
312 /// md.subscribe_order_book(&sber, SubscribeOpts { backpressure: Backpressure::Conflate, depth: Some(10) }).await?;
313 ///
314 /// let report = md.shutdown(ShutdownMode::Graceful { deadline: Duration::from_secs(1) }).await;
315 /// assert_eq!(report.unsubscribed + report.unsubscribe_failed, 1);
316 /// assert!(report.inflight_abandoned.is_empty());
317 /// # Ok(())
318 /// # }
319 /// ```
320 pub async fn shutdown(self, mode: ShutdownMode) -> ShutdownReport {
321 let mut unsubscribed = 0usize;
322 let mut unsubscribe_failed = 0usize;
323
324 if !matches!(mode, ShutdownMode::Immediate) {
325 let subs: Vec<_> = self.inner.subs.read().await.values().cloned().collect();
326 for sub in subs {
327 match self.inner.ws.unsubscribe(sub.ws.clone()).await {
328 Ok(()) => unsubscribed += 1,
329 Err(_) => unsubscribe_failed += 1,
330 }
331 }
332 }
333
334 if let ShutdownMode::Graceful { deadline } = mode {
335 drain_backlog(&self.inner.emitter, deadline).await;
336 }
337 self.inner.cancel.cancel();
338 let _ = self.pump.await;
339
340 ShutdownReport::new(unsubscribed, unsubscribe_failed, Vec::new())
341 }
342
343 async fn acquire(&self, method: &str) -> Result<(), Error> {
344 self.inner.limiter.acquire(method, brokapi_runtime::now())
345 }
346
347 async fn register_md_sub(
348 &self,
349 instrument: InstrumentId,
350 kind: MdSubKind,
351 core_kind: SubscriptionKind,
352 opts: SubscribeOpts,
353 ws: MarketDataSubscribe,
354 ) -> Result<SubscriptionId, Error> {
355 let id = self.inner.sub_ids.next();
356 self.inner
357 .emitter
358 .register(id, Some(instrument.clone()), opts.backpressure);
359 // The core attributes incoming data to this subscription (Envelope::sub, backpressure).
360 // It is told before the WS subscribe, so the success ack cannot overtake it.
361 self.set_core_sub(id, Some(core_kind)).await;
362 let notify = self.inner.core.lock().await.notifier();
363 if let Err(e) = self.inner.ws.subscribe(ws.clone()).await {
364 self.set_core_sub(id, None).await;
365 self.inner.emitter.unregister(id);
366 return Err(map_err(e));
367 }
368 // Returns when the broker has confirmed the subscription: what happens after that is delivered. A
369 // subscription that is refused or not confirmed in time is taken back, nothing stays half open.
370 let confirmed = wait_ack(
371 &self.inner.core,
372 notify,
373 self.inner.ack_timeout,
374 |c: &MdCore| match c.actual_state(id) {
375 Some(crate::core::SubState::Active) => AckState::Confirmed,
376 Some(crate::core::SubState::Failed) => {
377 AckState::Refused(c.failure(id).unwrap_or_default().to_string())
378 }
379 _ => AckState::Pending,
380 },
381 )
382 .await;
383 if let Err(e) = confirmed {
384 let _ = self.inner.ws.unsubscribe(ws).await;
385 self.set_core_sub(id, None).await;
386 self.inner.emitter.unregister(id);
387 return Err(e);
388 }
389 self.inner.subs.write().await.insert(
390 id,
391 MdSubscription {
392 id,
393 instrument,
394 kind,
395 opts,
396 ws,
397 },
398 );
399 Ok(id)
400 }
401
402 /// Adds (`Some`) or removes (`None`) a subscription in the core's desired set.
403 /// The core's own subscribe frames are dropped: `WsSession` sends them and restores
404 /// them after a reconnect.
405 async fn set_core_sub(&self, id: SubscriptionId, kind: Option<SubscriptionKind>) {
406 let mut core = self.inner.core.lock().await;
407 let mut subs: Vec<DesiredSubscription> = core
408 .desired()
409 .iter()
410 .filter(|s| s.id != id)
411 .cloned()
412 .collect();
413 if let Some(kind) = kind {
414 subs.push(DesiredSubscription { id, kind });
415 }
416 core.apply(&SubscriptionSet::with_subscriptions(subs));
417 while core.poll_transmit().is_some() {}
418 }
419}
420
421#[async_trait]
422impl brokapi_core::Connector for MarketDataConnector {
423 type Caps = brokapi_core::MarketDataCaps;
424
425 fn caps(&self) -> brokapi_core::MarketDataCaps {
426 MarketDataConnector::caps(self)
427 }
428
429 async fn shutdown(self, mode: brokapi_core::ShutdownMode) -> brokapi_core::ShutdownReport {
430 MarketDataConnector::shutdown(self, mode).await
431 }
432}
433
434#[async_trait]
435impl OrderBookSource for MarketDataConnector {
436 async fn order_book(&self, id: &InstrumentId, depth: u32) -> Result<OrderBook, Error> {
437 self.acquire("GetOrderBook").await?;
438 let depth = depth.min(MAX_DEPTH);
439 let req = OrderBookRequest {
440 ticker: ticker(id)?,
441 class_code: class_code(id)?,
442 depth,
443 };
444 let resp = self
445 .inner
446 .http
447 .get_order_book(&req)
448 .await
449 .map_err(map_err)?;
450 match map_rest_order_book(&resp, brokapi_runtime::now()) {
451 brokapi_core::Event::OrderBook(book) => Ok(book),
452 _ => Err(Error::Transport(brokapi_core::TransportError::Message(
453 "order book mapping failed".into(),
454 ))),
455 }
456 }
457
458 async fn subscribe_order_book(
459 &self,
460 id: &InstrumentId,
461 opts: SubscribeOpts,
462 ) -> Result<SubscriptionId, Error> {
463 let depth = opts.depth.unwrap_or(MAX_DEPTH).min(MAX_DEPTH);
464 let key = require_ticker_class(id)?;
465 let ws = MarketDataSubscribe::order_book(vec![key], depth);
466 self.register_md_sub(
467 id.clone(),
468 MdSubKind::OrderBook,
469 SubscriptionKind::OrderBook {
470 instrument: id.clone(),
471 depth,
472 },
473 opts,
474 ws,
475 )
476 .await
477 }
478
479 async fn unsubscribe(&self, sub: SubscriptionId) -> Result<(), Error> {
480 let record = self
481 .inner
482 .subs
483 .write()
484 .await
485 .remove(&sub)
486 .ok_or_else(|| Error::Rejected {
487 code: "unknown_subscription".into(),
488 message: format!("subscription {} is not active", sub.value()),
489 })?;
490 self.set_core_sub(sub, None).await;
491 self.inner.emitter.unregister(sub);
492 self.inner.ws.unsubscribe(record.ws).await.map_err(map_err)
493 }
494}
495
496#[async_trait]
497impl MarketDataSource for MarketDataConnector {
498 /// `POST /last-trades`: the tape of the current session. The API has no limit, so the newest
499 /// `limit` trades are kept here (oldest first). A record without a known side is dropped.
500 async fn last_trades(&self, id: &InstrumentId, limit: u32) -> Result<Vec<MarketTrade>, Error> {
501 self.acquire("GetLastTrades").await?;
502 let req = LastTradesRequest {
503 ticker: ticker(id)?,
504 class_code: class_code(id)?,
505 ..Default::default()
506 };
507 let mut resp = self
508 .inner
509 .http
510 .get_last_trades(&req)
511 .await
512 .map_err(map_err)?;
513 // Oldest first; records without a time keep their place (stable sort on a None key).
514 resp.sort_by_key(|t| crate::map::parse_broker_ts(t.date_time.as_deref()));
515 let now = brokapi_runtime::now();
516 let mut trades: Vec<MarketTrade> = resp
517 .iter()
518 .filter_map(|t| map_rest_last_trade(t, id, now))
519 .collect();
520 if trades.len() > limit as usize {
521 trades.drain(..trades.len() - limit as usize);
522 }
523 Ok(trades)
524 }
525
526 async fn subscribe_market_trades(
527 &self,
528 id: &InstrumentId,
529 opts: SubscribeOpts,
530 ) -> Result<SubscriptionId, Error> {
531 let key = require_ticker_class(id)?;
532 let ws = MarketDataSubscribe::last_trades(vec![key]);
533 self.register_md_sub(
534 id.clone(),
535 MdSubKind::MarketTrades,
536 SubscriptionKind::LastTrades {
537 instrument: id.clone(),
538 },
539 opts,
540 ws,
541 )
542 .await
543 }
544
545 async fn subscribe_quotes(
546 &self,
547 id: &InstrumentId,
548 opts: SubscribeOpts,
549 ) -> Result<SubscriptionId, Error> {
550 let key = require_ticker_class(id)?;
551 let ws = MarketDataSubscribe::quotes(vec![key]);
552 self.register_md_sub(
553 id.clone(),
554 MdSubKind::Quotes,
555 SubscriptionKind::Quotes {
556 instrument: id.clone(),
557 },
558 opts,
559 ws,
560 )
561 .await
562 }
563
564 async fn candles(
565 &self,
566 id: &InstrumentId,
567 tf: Timeframe,
568 range: TimeRange,
569 ) -> Result<Vec<Candle>, Error> {
570 let label = timeframe_label(tf).ok_or(Error::Unsupported {
571 capability: "candles_m10".into(),
572 })?;
573 self.acquire("GetCandles").await?;
574 let req = CandlesRequest {
575 ticker: ticker(id)?,
576 class_code: class_code(id)?,
577 start_date: nanos_to_datetime(range.from),
578 end_date: nanos_to_datetime(range.to),
579 time_frame: label.to_string(),
580 };
581 let resp = self.inner.http.get_candles(&req).await.map_err(map_err)?;
582 Ok(map_rest_candles(&resp, tf, brokapi_runtime::now()))
583 }
584
585 /// One request returns at most 1440 bars (BCS documentation of `candles-chart`), so the span is
586 /// 1440 bars of the timeframe. Not checked against the live API.
587 fn candles_max_span(&self, tf: Timeframe) -> Option<Duration> {
588 Some(Duration::from_secs(1440 * tf.seconds()))
589 }
590
591 async fn subscribe_candles(
592 &self,
593 id: &InstrumentId,
594 tf: Timeframe,
595 opts: SubscribeOpts,
596 ) -> Result<SubscriptionId, Error> {
597 let label = timeframe_label(tf).ok_or(Error::Unsupported {
598 capability: "candles_m10".into(),
599 })?;
600 let key = require_ticker_class(id)?;
601 let ws = MarketDataSubscribe::last_candle(vec![key], label);
602 self.register_md_sub(
603 id.clone(),
604 MdSubKind::Candles,
605 SubscriptionKind::LastCandle {
606 instrument: id.clone(),
607 timeframe: tf,
608 },
609 opts,
610 ws,
611 )
612 .await
613 }
614}
615
616#[async_trait]
617impl InstrumentSource for MarketDataConnector {
618 async fn find(&self, q: InstrumentQuery) -> Result<Vec<Instrument>, Error> {
619 let route = match &q {
620 InstrumentQuery::ByTicker { .. } => "GetInstrumentsByTickers",
621 InstrumentQuery::ByIsin(_) => "GetInstrumentsByIsins",
622 InstrumentQuery::ByKind(_) => "GetInstrumentsByType",
623 };
624 self.acquire(route).await?;
625 let infos = match q {
626 InstrumentQuery::ByTicker {
627 ticker,
628 exchange: _,
629 } => self
630 .inner
631 .http
632 .get_instruments_by_tickers(
633 &InstrumentsByTickersRequest {
634 tickers: vec![ticker.clone()],
635 },
636 None,
637 None,
638 )
639 .await
640 .map_err(map_err)?,
641 InstrumentQuery::ByIsin(isin) => self
642 .inner
643 .http
644 .get_instruments_by_isins(
645 &InstrumentsByIsinsRequest { isins: vec![isin] },
646 None,
647 None,
648 )
649 .await
650 .map_err(map_err)?,
651 InstrumentQuery::ByKind(kind) => {
652 // The docs: a page shorter than `size` ends the list, a full one means "ask for the
653 // next". Every page spends quota like a request of its own.
654 let ty = instrument_type(kind)?.to_string();
655 let mut all = Vec::new();
656 for page in 0..MAX_KIND_PAGES {
657 if page > 0 {
658 self.acquire(route).await?;
659 }
660 let batch = self
661 .inner
662 .http
663 .get_instruments_by_type(&InstrumentsByTypeRequest {
664 instrument_type: ty.clone(),
665 base_asset_ticker: None,
666 size: Some(KIND_PAGE_SIZE),
667 page: Some(page),
668 })
669 .await
670 .map_err(map_err)?;
671 let full = batch.len() >= KIND_PAGE_SIZE as usize;
672 all.extend(batch);
673 if !full {
674 break;
675 }
676 if page + 1 == MAX_KIND_PAGES {
677 return Err(Error::Transport(brokapi_core::TransportError::Message(
678 format!(
679 "the list of instruments of this kind is longer than {} pages: not returned in part",
680 MAX_KIND_PAGES
681 ),
682 )));
683 }
684 }
685 all
686 }
687 };
688 Ok(infos
689 .iter()
690 .map(|info| {
691 let class = if info.primary_board.is_empty() {
692 info.boards
693 .first()
694 .map(|b| b.class_code.clone())
695 .unwrap_or_default()
696 } else {
697 info.primary_board.clone()
698 };
699 map_instrument_info(info, &class)
700 })
701 .collect())
702 }
703
704 async fn get(&self, id: &InstrumentId) -> Result<Instrument, Error> {
705 let t = ticker(id)?;
706 let list = self
707 .find(InstrumentQuery::ByTicker {
708 ticker: t,
709 exchange: None,
710 })
711 .await?;
712 list.into_iter().next().ok_or_else(|| {
713 Error::Transport(brokapi_core::TransportError::Message(
714 "instrument not found".into(),
715 ))
716 })
717 }
718
719 async fn trading_params(&self, id: &InstrumentId) -> Result<TradingParams, Error> {
720 self.acquire("GetInstrumentsByTickers").await?;
721 let t = ticker(id)?;
722 let cc = class_code(id)?;
723 let infos = self
724 .inner
725 .http
726 .get_instruments_by_tickers(
727 &InstrumentsByTickersRequest {
728 tickers: vec![t.clone()],
729 },
730 None,
731 None,
732 )
733 .await
734 .map_err(map_err)?;
735 let info = infos.first().ok_or_else(|| {
736 Error::Transport(brokapi_core::TransportError::Message(
737 "instrument not found".into(),
738 ))
739 })?;
740 let snap = info.trading_params_for_board(&cc).ok_or_else(|| {
741 Error::Transport(brokapi_core::TransportError::Message(
742 "trading params unavailable".into(),
743 ))
744 })?;
745 Ok(map_trading_params(&snap))
746 }
747
748 async fn trading_status(&self, id: &InstrumentId) -> Result<TradingStatus, Error> {
749 self.acquire("GetTradingStatus").await?;
750 let req = TradingStatusRequest {
751 class_code: class_code(id)?,
752 };
753 let resp = self
754 .inner
755 .http
756 .get_trading_status(&req)
757 .await
758 .map_err(map_err)?;
759 Ok(map_trading_status(&ticker(id)?, &class_code(id)?, &resp))
760 }
761
762 /// `GET /trading-schedule/daily-schedule` needs the instrument (`ticker` and `classCode` are both
763 /// required), so BCS schedules are per instrument, like Finam's: `exchange` is `"TICKER@CLASSCODE"`
764 /// (`"SBER@TQBR"`). Anything else is a `Config` error. The answer is for one day, today: the
765 /// intervals are put on the day of `range.from` (an assumption, see `map/schedule.rs`).
766 async fn schedule(&self, exchange: &str, range: TimeRange) -> Result<Schedule, Error> {
767 let Some((t, c)) = exchange
768 .split_once('@')
769 .filter(|(t, c)| !t.is_empty() && !c.is_empty())
770 else {
771 return Err(Error::Config(brokapi_core::ConfigError::Message(
772 "BCS schedule is per instrument: pass \"TICKER@CLASSCODE\"".into(),
773 )));
774 };
775 self.acquire("GetDailySchedule").await?;
776 let req = DailyScheduleRequest {
777 class_code: c.to_string(),
778 ticker: t.to_string(),
779 };
780 let resp = self
781 .inner
782 .http
783 .get_daily_schedule(&req)
784 .await
785 .map_err(map_err)?;
786 let day = nanos_to_datetime(range.from).date_naive();
787 Ok(map_daily_schedule(
788 exchange,
789 day,
790 &resp,
791 brokapi_runtime::now(),
792 ))
793 }
794}
795
796/// `type` of `GET /instruments/by-type`: `STOCK`, `BONDS`, `FUTURES`, `ETF`, `MUTUAL_FUNDS`, `CURRENCY`
797/// (docs enum, checked 2026-10-19). Options need `baseAssetTicker`, which a kind query cannot carry, and
798/// `Other` has no BCS type.
799fn instrument_type(kind: brokapi_core::InstrumentKind) -> Result<&'static str, Error> {
800 use brokapi_core::InstrumentKind as K;
801 match kind {
802 K::Stock => Ok("STOCK"),
803 K::Bond => Ok("BONDS"),
804 K::Future => Ok("FUTURES"),
805 K::Etf => Ok("ETF"),
806 K::MutualFund => Ok("MUTUAL_FUNDS"),
807 K::Currency => Ok("CURRENCY"),
808 K::Perpetual => Err(Error::Unsupported {
809 capability: "find_perpetuals_by_kind".into(),
810 }),
811 K::Option => Err(Error::Unsupported {
812 capability: "find_options_by_kind".into(),
813 }),
814 K::Other => Err(Error::Unsupported {
815 capability: "find_other_by_kind".into(),
816 }),
817 }
818}