brokapi_bybit/
md.rs

1//! The market data connector: order books, trades, quotes, candles and the directory of instruments.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5use std::time::Duration;
6
7use async_trait::async_trait;
8use brokapi_bybit_protocol::http::client::BybitHttpClient;
9use brokapi_bybit_protocol::ws::session::{WsSession, WsStream};
10use brokapi_core::{
11    Candle, Envelope, Error, Instrument, InstrumentId, InstrumentKind, InstrumentQuery,
12    InstrumentSource, MarketDataCaps, MarketDataSource, MarketTrade, OrderBook, OrderBookSource,
13    Schedule, ShutdownMode, ShutdownReport, StreamingCaps, SubscribeOpts, SubscriptionId,
14    TimeRange, Timeframe, TradingParams, TradingStatus,
15};
16use brokapi_runtime::{BrokerLimiter, Emitter};
17use brokapi_sansio::SubscriptionKind;
18use tokio::sync::{Mutex, RwLock, mpsc};
19use tokio::task::JoinHandle;
20use tokio_util::sync::CancellationToken;
21
22use crate::config::MarketDataConfig;
23use crate::core::MdCore;
24use crate::error::{map_config_err, map_err};
25use crate::map::instrument::{instrument, kind_of, trading_params, trading_status};
26use crate::map::market::{candle, market_trade, order_book};
27use crate::map::unix_now_nanos;
28use crate::pump::{drain_backlog, run_pump};
29use crate::raw::BybitRaw;
30use crate::subs::{MdSubKind, MdSubscription, ReqIdGen, SubIdGen, wait_ack};
31use crate::util::{interval_label, nanos_to_ms, symbol_category};
32
33/// The deepest book the exchange gives for linear: 1000 levels a side.
34const MAX_DEPTH: u32 = 1000;
35/// The depths of the order book stream of linear.
36const STREAM_DEPTHS: [u32; 4] = [1, 50, 200, 1000];
37/// The default depth of a stream when the caller names none.
38const DEFAULT_DEPTH: u32 = 50;
39/// How many bars one request of the history answers at most.
40const KLINE_LIMIT: u32 = 1000;
41/// How many instruments a page of the directory holds at most.
42const PAGE_SIZE: u32 = 1000;
43const MAX_PAGES: u32 = 50;
44
45struct MdShared {
46    http: BybitHttpClient,
47    ws: WsSession,
48    limiter: Arc<BrokerLimiter>,
49    core: Arc<Mutex<MdCore>>,
50    emitter: Arc<Emitter>,
51    subs: RwLock<HashMap<SubscriptionId, MdSubscription>>,
52    sub_ids: SubIdGen,
53    req_ids: ReqIdGen,
54    cancel: CancellationToken,
55    ack_timeout: Duration,
56    category: String,
57}
58
59/// The market data connector of Bybit: implements [`OrderBookSource`], [`MarketDataSource`] and [`InstrumentSource`].
60///
61/// It is not tied to an account: the public data needs no key. Reads go by HTTP, streams come over one WebSocket
62/// session that reconnects on its own and restores its subscriptions. The streams are the public ones of the category
63/// (`wss://stream.bybit.com/v5/public/linear`): the demo contour has no public streams of its own, so a connector made
64/// from a demo config reads the same market as one made from a main config.
65///
66/// Instruments are named by the symbol and the category: `BTCUSDT@linear`
67/// ([`instrument_id`](crate::util::instrument_id)).
68///
69/// # Examples
70///
71/// ```
72/// # #[tokio::main(flavor = "current_thread")]
73/// # async fn main() -> Result<(), Box<dyn std::error::Error>> {
74/// # let server = brokapi_mocks::bybit::MockServer::start(Default::default()).await;
75/// # let cfg = brokapi_bybit_protocol::config::BybitConfig::for_mock("mock-key", "mock-secret", &server.http_base);
76/// use brokapi_bybit::limits::default_limits;
77/// use brokapi_bybit::util::instrument_id;
78/// use brokapi_bybit::{MarketDataConfig, MarketDataConnector};
79/// use brokapi_core::{OrderBook, OrderBookSource, ShutdownMode};
80/// use brokapi_runtime::BrokerLimiter;
81///
82/// let (tx, _events) = tokio::sync::mpsc::channel(256);
83/// let md = MarketDataConnector::start(MarketDataConfig::new(cfg, "bybit-md"), BrokerLimiter::new(&default_limits()), tx).await?;
84/// let OrderBook::Snapshot { bids, asks, .. } = md.order_book(&instrument_id("MOCKUSDT", "linear"), 5).await? else {
85///     panic!("a read answers a snapshot");
86/// };
87/// assert!(bids[0].price < asks[0].price);
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: reads the clock of the exchange and opens the WebSocket session in the background.
99    ///
100    /// # Errors
101    ///
102    /// [`Error::Config`] when the client cannot be built; [`Error::Transport`] when the exchange cannot be reached.
103    pub async fn start(
104        cfg: MarketDataConfig,
105        limiter: Arc<BrokerLimiter>,
106        out: mpsc::Sender<Envelope>,
107    ) -> Result<Self, Error> {
108        let http = BybitHttpClient::new(cfg.bybit.clone()).map_err(map_config_err)?;
109        http.sync_time().await.map_err(map_err)?;
110
111        let (event_tx, event_rx) = kanal::bounded_async(512);
112        let ws = WsSession::spawn(http.clone(), WsStream::Public, event_tx, cfg.ws_options());
113        let cancel = CancellationToken::new();
114        let core = Arc::new(Mutex::new(MdCore::new()));
115        let emitter = Emitter::new(Arc::clone(&cfg.source), out);
116
117        let inner = Arc::new(MdShared {
118            http,
119            ws,
120            limiter,
121            core: Arc::clone(&core),
122            emitter: Arc::clone(&emitter),
123            subs: RwLock::new(HashMap::new()),
124            sub_ids: SubIdGen::new(0),
125            req_ids: ReqIdGen::default(),
126            cancel: cancel.clone(),
127            ack_timeout: cfg.subscribe_ack_timeout,
128            category: cfg.bybit.category.as_str().to_string(),
129        });
130        let hook = Arc::clone(&inner);
131        let pump = tokio::spawn(run_pump(core, emitter, event_rx, cancel, move || {
132            let hook = Arc::clone(&hook);
133            async move { hook.ask_for_lost_books().await }
134        }));
135        Ok(Self { inner, pump })
136    }
137
138    /// Direct access to any API method with this connector's key and rate limiter. See [`BybitRaw`].
139    pub fn raw(&self) -> BybitRaw {
140        BybitRaw::new(self.inner.http.clone(), Arc::clone(&self.inner.limiter))
141    }
142
143    /// What the connector can do: books up to 1000 levels (the stream sends the whole book after every change, not
144    /// increments), and all four streams.
145    pub fn caps(&self) -> MarketDataCaps {
146        MarketDataCaps::new(MAX_DEPTH, false, StreamingCaps::new(true, true, true, true))
147    }
148
149    /// Stops the connector. Unless the mode is `Immediate` it cancels the streams first; `Graceful` also lets the bot's
150    /// channel take what is still buffered.
151    pub async fn shutdown(self, mode: ShutdownMode) -> ShutdownReport {
152        let mut unsubscribed = 0usize;
153        let mut unsubscribe_failed = 0usize;
154        if !matches!(mode, ShutdownMode::Immediate) {
155            let mut topics: Vec<String> = self
156                .inner
157                .subs
158                .read()
159                .await
160                .values()
161                .map(|s| s.topic.clone())
162                .collect();
163            topics.sort();
164            topics.dedup();
165            for topic in topics {
166                match self.inner.ws.unsubscribe(&topic).await {
167                    Ok(()) => unsubscribed += 1,
168                    Err(_) => unsubscribe_failed += 1,
169                }
170            }
171        }
172        if let ShutdownMode::Graceful { deadline } = mode {
173            drain_backlog(&self.inner.emitter, deadline).await;
174        }
175        self.inner.cancel.cancel();
176        let _ = self.pump.await;
177        ShutdownReport::new(unsubscribed, unsubscribe_failed, Vec::new())
178    }
179
180    fn acquire(&self, method: &str) -> Result<(), Error> {
181        self.inner.limiter.acquire(method, brokapi_runtime::now())
182    }
183
184    fn symbol(&self, id: &InstrumentId) -> Result<String, Error> {
185        symbol_category(id, &self.inner.category)
186    }
187
188    async fn register_md_sub(
189        &self,
190        instrument: InstrumentId,
191        kind: MdSubKind,
192        core_kind: SubscriptionKind,
193        opts: SubscribeOpts,
194        topic: String,
195    ) -> Result<SubscriptionId, Error> {
196        let id = self.inner.sub_ids.next();
197        let req = self.inner.req_ids.next();
198        self.inner
199            .emitter
200            .register(id, Some(instrument.clone()), opts.backpressure);
201        // The core attributes incoming data to this subscription. It is told before the request goes out, so the
202        // answer cannot overtake it.
203        let (first_on_topic, notify) = {
204            let mut core = self.inner.core.lock().await;
205            let first = !core.has_topic(&topic);
206            core.register(id, core_kind, &topic, Some(&req));
207            (first, core.notifier())
208        };
209        let sent = if first_on_topic {
210            self.inner.ws.subscribe(&req, &topic).await
211        } else if kind == MdSubKind::OrderBook {
212            // The topic is already open for someone else: this book has no snapshot yet, so the exchange is asked
213            // for the topic again, which sends one to all the listeners.
214            self.inner.ws.resubscribe(&req, &topic).await
215        } else {
216            Ok(())
217        };
218        if let Err(e) = sent {
219            self.drop_sub(id).await;
220            return Err(map_err(e));
221        }
222        // Returns when the exchange has confirmed the stream: what happens after that is delivered.
223        let confirmed = wait_ack(
224            &self.inner.core,
225            notify,
226            self.inner.ack_timeout,
227            |c: &MdCore| c.sub_state(id),
228        )
229        .await;
230        if let Err(e) = confirmed {
231            self.drop_sub(id).await;
232            return Err(e);
233        }
234        self.inner.subs.write().await.insert(
235            id,
236            MdSubscription {
237                id,
238                instrument,
239                kind,
240                opts,
241                topic,
242            },
243        );
244        Ok(id)
245    }
246
247    /// Takes a subscription back: the core forgets it, the exchange is told when nobody else listens to the topic.
248    async fn drop_sub(&self, id: SubscriptionId) {
249        let orphans = self.inner.core.lock().await.forget(id);
250        self.inner.emitter.unregister(id);
251        for topic in orphans {
252            let _ = self.inner.ws.unsubscribe(&topic).await;
253        }
254    }
255}
256
257impl MdShared {
258    /// Asks the exchange for a new snapshot of every order book topic that lost a delta.
259    async fn ask_for_lost_books(&self) {
260        let topics = self.core.lock().await.take_gaps();
261        for topic in topics {
262            let req = self.req_ids.next();
263            self.core.lock().await.add_request(&req, &topic);
264            let _ = self.ws.resubscribe(&req, &topic).await;
265        }
266    }
267}
268
269#[async_trait]
270impl brokapi_core::Connector for MarketDataConnector {
271    type Caps = brokapi_core::MarketDataCaps;
272
273    fn caps(&self) -> brokapi_core::MarketDataCaps {
274        MarketDataConnector::caps(self)
275    }
276
277    async fn shutdown(self, mode: brokapi_core::ShutdownMode) -> brokapi_core::ShutdownReport {
278        MarketDataConnector::shutdown(self, mode).await
279    }
280}
281
282#[async_trait]
283impl OrderBookSource for MarketDataConnector {
284    async fn order_book(&self, id: &InstrumentId, depth: u32) -> Result<OrderBook, Error> {
285        let symbol = self.symbol(id)?;
286        self.acquire("GetOrderBook")?;
287        let view = self
288            .inner
289            .http
290            .get_order_book(
291                &self.inner.category,
292                &symbol,
293                Some(depth.clamp(1, MAX_DEPTH)),
294            )
295            .await
296            .map_err(map_err)?;
297        Ok(order_book(
298            &view.bids,
299            &view.asks,
300            view.ts_ms,
301            id,
302            brokapi_runtime::now(),
303        ))
304    }
305
306    async fn subscribe_order_book(
307        &self,
308        id: &InstrumentId,
309        opts: SubscribeOpts,
310    ) -> Result<SubscriptionId, Error> {
311        let symbol = self.symbol(id)?;
312        let depth = opts.depth.unwrap_or(DEFAULT_DEPTH).clamp(1, MAX_DEPTH);
313        // The stream has fixed depths: the smallest one that holds what was asked for, cut to `depth` by the core.
314        let stream_depth = STREAM_DEPTHS
315            .iter()
316            .copied()
317            .find(|d| *d >= depth)
318            .unwrap_or(MAX_DEPTH);
319        let topic = format!("orderbook.{stream_depth}.{symbol}");
320        self.register_md_sub(
321            id.clone(),
322            MdSubKind::OrderBook,
323            SubscriptionKind::OrderBook {
324                instrument: id.clone(),
325                depth,
326            },
327            opts,
328            topic,
329        )
330        .await
331    }
332
333    async fn unsubscribe(&self, sub: SubscriptionId) -> Result<(), Error> {
334        let record = self
335            .inner
336            .subs
337            .write()
338            .await
339            .remove(&sub)
340            .ok_or_else(|| Error::Rejected {
341                code: "unknown_subscription".into(),
342                message: format!("subscription {} is not active", sub.value()),
343            })?;
344        let orphans = self.inner.core.lock().await.forget(sub);
345        self.inner.emitter.unregister(sub);
346        let mut result = Ok(());
347        for topic in orphans {
348            if let Err(e) = self.inner.ws.unsubscribe(&topic).await {
349                result = Err(map_err(e));
350            }
351        }
352        let _ = record;
353        result
354    }
355}
356
357#[async_trait]
358impl MarketDataSource for MarketDataConnector {
359    async fn last_trades(&self, id: &InstrumentId, limit: u32) -> Result<Vec<MarketTrade>, Error> {
360        let symbol = self.symbol(id)?;
361        self.acquire("GetLastTrades")?;
362        let records = self
363            .inner
364            .http
365            .get_recent_trades(&self.inner.category, &symbol, limit.clamp(1, 1000))
366            .await
367            .map_err(map_err)?;
368        let now = brokapi_runtime::now();
369        // Newest first from the exchange; the shared interface wants the oldest first.
370        let mut trades: Vec<MarketTrade> = records
371            .iter()
372            .rev()
373            .filter_map(|r| market_trade(r, id, now))
374            .collect();
375        if trades.len() > limit as usize {
376            trades.drain(..trades.len() - limit as usize);
377        }
378        Ok(trades)
379    }
380
381    async fn subscribe_market_trades(
382        &self,
383        id: &InstrumentId,
384        opts: SubscribeOpts,
385    ) -> Result<SubscriptionId, Error> {
386        let symbol = self.symbol(id)?;
387        self.register_md_sub(
388            id.clone(),
389            MdSubKind::MarketTrades,
390            SubscriptionKind::LastTrades {
391                instrument: id.clone(),
392            },
393            opts,
394            format!("publicTrade.{symbol}"),
395        )
396        .await
397    }
398
399    async fn subscribe_quotes(
400        &self,
401        id: &InstrumentId,
402        opts: SubscribeOpts,
403    ) -> Result<SubscriptionId, Error> {
404        let symbol = self.symbol(id)?;
405        // The best level of the book is the quote: the level 1 stream pushes it within 10 ms of a change.
406        self.register_md_sub(
407            id.clone(),
408            MdSubKind::Quotes,
409            SubscriptionKind::Quotes {
410                instrument: id.clone(),
411            },
412            opts,
413            format!("orderbook.1.{symbol}"),
414        )
415        .await
416    }
417
418    async fn candles(
419        &self,
420        id: &InstrumentId,
421        tf: Timeframe,
422        range: TimeRange,
423    ) -> Result<Vec<Candle>, Error> {
424        let symbol = self.symbol(id)?;
425        let interval = interval_label(tf).ok_or_else(|| Error::Unsupported {
426            capability: "candles_m10".into(),
427        })?;
428        let span = KLINE_LIMIT as u64 * tf.seconds();
429        let asked = (range.to.as_nanos().saturating_sub(range.from.as_nanos()) / 1_000_000_000)
430            .max(0) as u64;
431        if asked > span {
432            return Err(Error::Rejected {
433                code: "range_too_long".into(),
434                message: format!(
435                    "one request covers {KLINE_LIMIT} bars ({span} seconds of this timeframe) and the range is {asked} seconds: read it in windows (brokapi_runtime::history does)"
436                ),
437            });
438        }
439        self.acquire("GetCandles")?;
440        let rows = self
441            .inner
442            .http
443            .get_klines(
444                &self.inner.category,
445                &symbol,
446                interval,
447                Some(nanos_to_ms(range.from)),
448                Some(nanos_to_ms(range.to)),
449                KLINE_LIMIT,
450            )
451            .await
452            .map_err(map_err)?;
453        let now = brokapi_runtime::now();
454        Ok(rows
455            .iter()
456            // The window is half open: a bar that starts at `to` belongs to the next one.
457            .filter(|k| k.start_ms < nanos_to_ms(range.to))
458            .map(|k| candle(k, id, tf, now))
459            .collect())
460    }
461
462    fn candles_max_span(&self, tf: Timeframe) -> Option<Duration> {
463        interval_label(tf)?;
464        Some(Duration::from_secs(KLINE_LIMIT as u64 * tf.seconds()))
465    }
466
467    async fn subscribe_candles(
468        &self,
469        id: &InstrumentId,
470        tf: Timeframe,
471        opts: SubscribeOpts,
472    ) -> Result<SubscriptionId, Error> {
473        let symbol = self.symbol(id)?;
474        let interval = interval_label(tf).ok_or_else(|| Error::Unsupported {
475            capability: "candles_m10".into(),
476        })?;
477        self.register_md_sub(
478            id.clone(),
479            MdSubKind::Candles,
480            SubscriptionKind::LastCandle {
481                instrument: id.clone(),
482                timeframe: tf,
483            },
484            opts,
485            format!("kline.{interval}.{symbol}"),
486        )
487        .await
488    }
489}
490
491#[async_trait]
492impl InstrumentSource for MarketDataConnector {
493    async fn find(&self, q: InstrumentQuery) -> Result<Vec<Instrument>, Error> {
494        match q {
495            InstrumentQuery::ByTicker { ticker, exchange } => {
496                // The exchange is one: BYBIT, or the category written the way the key does.
497                if exchange.as_deref().is_some_and(|e| {
498                    !e.eq_ignore_ascii_case("BYBIT")
499                        && !e.eq_ignore_ascii_case(&self.inner.category)
500                }) {
501                    return Ok(Vec::new());
502                }
503                self.acquire("GetInstruments")?;
504                match self
505                    .inner
506                    .http
507                    .get_instrument(&self.inner.category, &ticker.to_ascii_uppercase())
508                    .await
509                {
510                    Ok(info) => Ok(vec![instrument(&info, &self.inner.category)]),
511                    Err(e) => match map_err(e) {
512                        e if e.is_not_found() => Ok(Vec::new()),
513                        e => Err(e),
514                    },
515                }
516            }
517            InstrumentQuery::ByIsin(_) => Err(Error::Unsupported {
518                capability: "find_by_isin".into(),
519            }),
520            InstrumentQuery::ByKind(kind) => self.find_by_kind(kind).await,
521        }
522    }
523
524    async fn get(&self, id: &InstrumentId) -> Result<Instrument, Error> {
525        let symbol = self.symbol(id)?;
526        self.acquire("GetInstruments")?;
527        let info = self
528            .inner
529            .http
530            .get_instrument(&self.inner.category, &symbol)
531            .await
532            .map_err(map_err)?;
533        Ok(instrument(&info, &self.inner.category))
534    }
535
536    async fn trading_params(&self, id: &InstrumentId) -> Result<TradingParams, Error> {
537        let symbol = self.symbol(id)?;
538        self.acquire("GetInstruments")?;
539        let info = self
540            .inner
541            .http
542            .get_instrument(&self.inner.category, &symbol)
543            .await
544            .map_err(map_err)?;
545        Ok(trading_params(&info, &self.inner.category))
546    }
547
548    async fn trading_status(&self, id: &InstrumentId) -> Result<TradingStatus, Error> {
549        let symbol = self.symbol(id)?;
550        self.acquire("GetInstruments")?;
551        let info = self
552            .inner
553            .http
554            .get_instrument(&self.inner.category, &symbol)
555            .await
556            .map_err(map_err)?;
557        Ok(trading_status(
558            &info,
559            &self.inner.category,
560            Some(unix_now_nanos()),
561        ))
562    }
563
564    async fn schedule(&self, _exchange: &str, _range: TimeRange) -> Result<Schedule, Error> {
565        // A crypto exchange trades round the clock: there is no timetable of sessions.
566        Err(Error::Unsupported {
567            capability: "schedule".into(),
568        })
569    }
570}
571
572impl MarketDataConnector {
573    async fn find_by_kind(&self, kind: InstrumentKind) -> Result<Vec<Instrument>, Error> {
574        match kind {
575            InstrumentKind::Perpetual | InstrumentKind::Future => {}
576            InstrumentKind::Other => {
577                return Err(Error::Unsupported {
578                    capability: "find_other_by_kind".into(),
579                });
580            }
581            // Category linear holds contracts only.
582            _ => return Ok(Vec::new()),
583        }
584        let mut all = Vec::new();
585        let mut cursor: Option<String> = None;
586        for page in 0..MAX_PAGES {
587            if page > 0 {
588                self.acquire("GetInstruments")?;
589            }
590            let batch = self
591                .inner
592                .http
593                .get_instruments_page(&self.inner.category, PAGE_SIZE, cursor.as_deref())
594                .await
595                .map_err(map_err)?;
596            all.extend(
597                batch
598                    .list
599                    .iter()
600                    .filter(|i| kind_of(&i.contract_type) == kind)
601                    .map(|i| instrument(i, &self.inner.category)),
602            );
603            if batch.next_page_cursor.is_empty() {
604                return Ok(all);
605            }
606            cursor = Some(batch.next_page_cursor);
607        }
608        Err(Error::Transport(brokapi_core::TransportError::Message(
609            format!(
610                "the list of instruments is longer than {MAX_PAGES} pages: not returned in part"
611            ),
612        )))
613    }
614}
brokapi_bybit/md.rs

brokapi_bybit/
md.rs
