brokapi_bcs/core/
md.rs

1//! Market-data sans-io core.
2
3use std::collections::{HashMap, VecDeque};
4use std::sync::Arc;
5use std::time::{Duration, Instant};
6
7use brokapi_bcs_protocol::models::instrument::InstrumentKey as BcsInstrumentKey;
8use brokapi_bcs_protocol::models::messages::IncomingMessage;
9use brokapi_bcs_protocol::ws::client::BcsWsChannel;
10use brokapi_bcs_protocol::ws::models::{MarketDataMessage, MarketDataSubscribe};
11use brokapi_bcs_protocol::ws::parse_incoming;
12use brokapi_core::{Event, ResyncReason, SubscriptionId, Timeframe};
13use brokapi_sansio::{
14    BrokerCore, CoreEvent, DesiredSubscription, EventSink, Outbound, SubscriptionKind,
15    SubscriptionSet,
16};
17use tokio::sync::Notify;
18
19use crate::map::market::{map_ws_market_data, parse_subscribe_error};
20
21/// The state of one subscription as the core sees it.
22#[derive(Debug, Clone, Copy, PartialEq, Eq)]
23pub enum SubState {
24    /// The subscribe message is queued or sent, and the broker has not confirmed it yet. Also the
25    /// state after a reconnect, until the broker confirms again. Data that arrives for a pending
26    /// subscription marks it active, because data can overtake the confirmation.
27    Pending,
28    /// The broker confirmed the subscription, or data for it has arrived.
29    Active,
30    /// The broker answered with an error while the subscription was pending. The core does not
31    /// retry. The connector reports it to the caller.
32    Failed,
33}
34
35/// Sans-io market data core of the BCS market data WebSocket.
36///
37/// It turns inbound frames into [`Event`]s (order books, quotes, trades, candles), tracks which
38/// subscriptions the broker has confirmed, and after a reconnect tells the consumer that the order
39/// books are void (`Event::Resync` with [`ResyncReason::SubscriptionReset`]). It does no I/O.
40///
41/// # Examples
42///
43/// Ask for an order book subscription and read the message the core wants to send:
44///
45/// ```
46/// use std::time::Instant;
47/// use brokapi_bcs::{MdCore, core::SubState};
48/// use brokapi_core::{InstrumentId, InstrumentKey, NativeId, SubscriptionId};
49/// use brokapi_sansio::{
50///     BrokerCore, DesiredSubscription, EventSink, Outbound, SubscriptionKind, SubscriptionSet,
51/// };
52///
53/// let sber = InstrumentId::new(
54///     InstrumentKey::new("SBER", "TQBR"),
55///     NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
56/// );
57/// let sub = SubscriptionId::new(1);
58/// let mut core = MdCore::new();
59/// core.apply(&SubscriptionSet::with_subscriptions(vec![DesiredSubscription {
60///     id: sub,
61///     kind: SubscriptionKind::OrderBook { instrument: sber, depth: 10 },
62/// }]));
63///
64/// let Some(Outbound::WsText(text)) = core.poll_transmit() else { panic!("a subscribe message") };
65/// assert!(text.contains("\"subscribeType\":0"));
66/// // Waiting for the broker's confirmation.
67/// assert_eq!(core.actual_state(sub), Some(SubState::Pending));
68///
69/// // A subscribe error from the broker fails the subscriptions that wait.
70/// let mut out = EventSink::new();
71/// core.feed_raw(r#"{"errors":[{"message":"unknown instrument"}]}"#, Instant::now(), &mut out);
72/// assert_eq!(core.actual_state(sub), Some(SubState::Failed));
73/// ```
74#[derive(Debug)]
75pub struct MdCore {
76    desired: SubscriptionSet,
77    actual: HashMap<SubscriptionId, SubState>,
78    sub_by_id: HashMap<SubscriptionId, DesiredSubscription>,
79    outbound: VecDeque<Outbound>,
80    book_seq: HashMap<SubscriptionId, u64>,
81    heartbeat_deadline: Option<Instant>,
82    /// How many times the WebSocket session has connected (the first connect included).
83    connections: u32,
84    /// Why a subscription failed, in the words of the broker.
85    failures: HashMap<SubscriptionId, String>,
86    /// Woken whenever a subscription changes state (confirmed or refused): the connector waits on it in
87    /// `subscribe_*` until the broker has answered.
88    ack: Arc<Notify>,
89}
90
91impl Default for MdCore {
92    fn default() -> Self {
93        Self::new()
94    }
95}
96
97impl MdCore {
98    /// Creates a core with no subscriptions.
99    pub fn new() -> Self {
100        Self {
101            desired: SubscriptionSet::new(),
102            actual: HashMap::new(),
103            sub_by_id: HashMap::new(),
104            outbound: VecDeque::new(),
105            book_seq: HashMap::new(),
106            heartbeat_deadline: None,
107            connections: 0,
108            failures: HashMap::new(),
109            ack: Arc::new(Notify::new()),
110        }
111    }
112
113    /// The signal that is given whenever a subscription is confirmed or refused. A connector waits on it
114    /// (with [`brokapi_runtime::wait_ack`]) to hold `subscribe_*` until the broker has answered.
115    pub fn notifier(&self) -> Arc<Notify> {
116        Arc::clone(&self.ack)
117    }
118
119    /// Why subscription `id` failed, in the words of the broker, if it did.
120    pub fn failure(&self, id: SubscriptionId) -> Option<&str> {
121        self.failures.get(&id).map(String::as_str)
122    }
123
124    /// Subscriptions the connector asked for (input of the last `apply`).
125    pub fn desired(&self) -> &SubscriptionSet {
126        &self.desired
127    }
128
129    /// The state of subscription `id`, or `None` if the core does not know it (never applied, or
130    /// already unsubscribed).
131    pub fn actual_state(&self, id: SubscriptionId) -> Option<SubState> {
132        self.actual.get(&id).copied()
133    }
134
135    /// The session connected again. Its subscriptions are restored by the session itself, so nothing is
136    /// sent here, but every order book is void: the stream starts over with a new snapshot, and the
137    /// consumer is told so with `Resync { SubscriptionReset }` (before the new snapshot, which comes after
138    /// the restore). The subscriptions wait for their acknowledgements again.
139    fn on_reconnected(&mut self, out: &mut EventSink) {
140        self.connections += 1;
141        if self.connections < 2 {
142            return;
143        }
144        for sub in self.desired.subs() {
145            if self.actual.contains_key(&sub.id) {
146                self.actual.insert(sub.id, SubState::Pending);
147            }
148            if let SubscriptionKind::OrderBook { instrument, .. } = &sub.kind {
149                self.book_seq.remove(&sub.id);
150                out.push(CoreEvent {
151                    sub: Some(sub.id),
152                    payload: Event::Resync {
153                        sub: sub.id,
154                        instrument: instrument.clone(),
155                        reason: ResyncReason::SubscriptionReset,
156                    },
157                });
158            }
159        }
160    }
161
162    fn queue_subscribe(&mut self, sub: &DesiredSubscription) {
163        let msg = match &sub.kind {
164            SubscriptionKind::OrderBook { instrument, depth } => {
165                let key = bcs_key(instrument);
166                MarketDataSubscribe::order_book(vec![key], *depth)
167            }
168            SubscriptionKind::Quotes { instrument } => {
169                MarketDataSubscribe::quotes(vec![bcs_key(instrument)])
170            }
171            SubscriptionKind::LastTrades { instrument } => {
172                MarketDataSubscribe::last_trades(vec![bcs_key(instrument)])
173            }
174            SubscriptionKind::LastCandle {
175                instrument,
176                timeframe,
177            } => MarketDataSubscribe::last_candle(
178                vec![bcs_key(instrument)],
179                timeframe_label(*timeframe),
180            ),
181            SubscriptionKind::OrderState
182            | SubscriptionKind::Portfolio
183            | SubscriptionKind::Executions => return,
184        };
185        self.outbound
186            .push_back(Outbound::WsText(serde_json::to_string(&msg).unwrap()));
187        self.actual.insert(sub.id, SubState::Pending);
188        self.sub_by_id.insert(sub.id, sub.clone());
189    }
190
191    fn queue_unsubscribe(&mut self, sub: &DesiredSubscription) {
192        let msg = match &sub.kind {
193            SubscriptionKind::OrderBook { instrument, depth } => {
194                MarketDataSubscribe::unsubscribe_order_book(vec![bcs_key(instrument)], *depth)
195            }
196            SubscriptionKind::Quotes { instrument } => {
197                MarketDataSubscribe::unsubscribe_quotes(vec![bcs_key(instrument)])
198            }
199            SubscriptionKind::LastTrades { instrument } => {
200                MarketDataSubscribe::unsubscribe_last_trades(vec![bcs_key(instrument)])
201            }
202            SubscriptionKind::LastCandle {
203                instrument,
204                timeframe,
205            } => MarketDataSubscribe::unsubscribe_last_candle(
206                vec![bcs_key(instrument)],
207                timeframe_label(*timeframe),
208            ),
209            SubscriptionKind::OrderState
210            | SubscriptionKind::Portfolio
211            | SubscriptionKind::Executions => return,
212        };
213        self.outbound
214            .push_back(Outbound::WsText(serde_json::to_string(&msg).unwrap()));
215        self.actual.remove(&sub.id);
216        self.failures.remove(&sub.id);
217        self.sub_by_id.remove(&sub.id);
218        self.book_seq.remove(&sub.id);
219    }
220
221    fn handle_market_data(&mut self, md: MarketDataMessage, now: Instant, out: &mut EventSink) {
222        match &md {
223            MarketDataMessage::OrderBookSuccess(_)
224            | MarketDataMessage::QuotesSuccess(_)
225            | MarketDataMessage::LastTradesSuccess(_)
226            | MarketDataMessage::LastCandleSuccess(_) => {
227                if let Some(sub_id) = self.match_success(&md) {
228                    self.actual.insert(sub_id, SubState::Active);
229                    self.ack.notify_waiters();
230                }
231                return;
232            }
233            MarketDataMessage::ErrorResponse(err) => {
234                let reason = err
235                    .errors
236                    .first()
237                    .map(|e| e.message.clone())
238                    .or_else(|| err.error.as_ref().map(|e| e.message.clone()))
239                    .unwrap_or_else(|| "subscription error".into());
240                self.fail_pending(&reason);
241                return;
242            }
243            _ => {}
244        }
245
246        if let Some(mapped) = map_ws_market_data(&md, now, None) {
247            let sub_id = self.match_data_sub(&md);
248            // Data that comes for a pending subscription proves it: the confirmation can be late or lost.
249            if let Some(id) = sub_id
250                && self.actual.get(&id) == Some(&SubState::Pending)
251            {
252                self.actual.insert(id, SubState::Active);
253                self.ack.notify_waiters();
254            }
255            let payload = mapped.event;
256            if let (Some(sub), Event::OrderBook(_)) = (sub_id, &payload) {
257                let seq = self.book_seq.entry(sub).or_insert(0);
258                *seq += 1;
259            }
260            out.push(CoreEvent {
261                sub: sub_id.or(mapped.sub),
262                payload,
263            });
264        }
265    }
266
267    fn match_success(&self, md: &MarketDataMessage) -> Option<SubscriptionId> {
268        for sub in self.desired.subs() {
269            if self.actual.get(&sub.id) != Some(&SubState::Pending) {
270                continue;
271            }
272            let matches = matches!(
273                (&sub.kind, md),
274                (
275                    SubscriptionKind::OrderBook { .. },
276                    MarketDataMessage::OrderBookSuccess(_)
277                ) | (
278                    SubscriptionKind::Quotes { .. },
279                    MarketDataMessage::QuotesSuccess(_)
280                ) | (
281                    SubscriptionKind::LastTrades { .. },
282                    MarketDataMessage::LastTradesSuccess(_)
283                ) | (
284                    SubscriptionKind::LastCandle { .. },
285                    MarketDataMessage::LastCandleSuccess(_)
286                )
287            );
288            if matches {
289                return Some(sub.id);
290            }
291        }
292        None
293    }
294
295    fn match_data_sub(&self, md: &MarketDataMessage) -> Option<SubscriptionId> {
296        for sub in self.desired.subs() {
297            // Data can overtake the success ack (or the ack can be lost on reconnect):
298            // a pending subscription that receives data is effectively active.
299            if !matches!(
300                self.actual.get(&sub.id),
301                Some(SubState::Active | SubState::Pending)
302            ) {
303                continue;
304            }
305            let matches = match (&sub.kind, md) {
306                (
307                    SubscriptionKind::OrderBook { instrument, .. },
308                    MarketDataMessage::OrderBook(d),
309                ) => same_instrument(instrument, &d.ticker, &d.class_code),
310                (SubscriptionKind::Quotes { instrument }, MarketDataMessage::Quotes(d)) => {
311                    same_instrument(instrument, &d.ticker, &d.class_code)
312                }
313                (SubscriptionKind::LastTrades { instrument }, MarketDataMessage::LastTrades(d)) => {
314                    same_instrument(instrument, &d.ticker, &d.class_code)
315                }
316                (
317                    SubscriptionKind::LastCandle { instrument, .. },
318                    MarketDataMessage::LastCandle(d),
319                ) => d
320                    .ticker
321                    .as_deref()
322                    .zip(d.class_code.as_deref())
323                    .is_some_and(|(t, c)| same_instrument(instrument, t, c)),
324                _ => false,
325            };
326            if matches {
327                return Some(sub.id);
328            }
329        }
330        None
331    }
332
333    /// Fails the pending subscriptions the error frame is about. The frame names no subscription, only a
334    /// code and a text. So: if exactly one subscription is pending, it is that one. If several are, the
335    /// ones whose ticker the text mentions. If the text names none of them (or there is no text to read),
336    /// the error cannot be placed and all pending subscriptions fail, as the safe reading.
337    fn fail_pending(&mut self, reason: &str) {
338        let pending: Vec<SubscriptionId> = self
339            .actual
340            .iter()
341            .filter(|(_, st)| **st == SubState::Pending)
342            .map(|(id, _)| *id)
343            .collect();
344        let text = reason.to_uppercase();
345        let mentioned: Vec<SubscriptionId> = self
346            .desired
347            .subs()
348            .iter()
349            .filter(|s| pending.contains(&s.id))
350            .filter(|s| {
351                let symbol = match &s.kind {
352                    SubscriptionKind::OrderBook { instrument, .. }
353                    | SubscriptionKind::Quotes { instrument }
354                    | SubscriptionKind::LastTrades { instrument }
355                    | SubscriptionKind::LastCandle { instrument, .. } => &instrument.key.symbol,
356                    _ => return false,
357                };
358                !symbol.is_empty() && text.contains(&symbol.to_uppercase())
359            })
360            .map(|s| s.id)
361            .collect();
362        let failed: Vec<SubscriptionId> = if pending.len() == 1 || mentioned.is_empty() {
363            pending
364        } else {
365            mentioned
366        };
367        for id in failed {
368            self.actual.insert(id, SubState::Failed);
369            self.failures.insert(id, reason.to_string());
370        }
371        self.ack.notify_waiters();
372    }
373
374    /// Feeds one raw text frame of the market data WebSocket.
375    ///
376    /// A frame with an `errors` array is a subscribe error: the pending subscription it is about becomes
377    /// [`SubState::Failed`] (all pending ones if the frame does not say which, see `fail_pending`) and no
378    /// event is pushed. Any other frame is parsed and handled like
379    /// [`on_frame`](BrokerCore::on_frame) does. A frame that cannot be parsed is skipped, there is
380    /// no error to return.
381    pub fn feed_raw(&mut self, text: &str, now: Instant, out: &mut EventSink) {
382        if let Some(reason) = parse_subscribe_error(text) {
383            self.fail_pending(&reason);
384            return;
385        }
386        let frame = parse_incoming(text, BcsWsChannel::MarketData);
387        self.on_frame(frame, now, out);
388    }
389}
390
391impl BrokerCore for MdCore {
392    type Frame = IncomingMessage;
393
394    fn on_frame(&mut self, frame: Self::Frame, now: Instant, out: &mut EventSink) {
395        self.heartbeat_deadline = Some(now + Duration::from_secs(30));
396        match frame {
397            IncomingMessage::MarketData(md) => self.handle_market_data(md, now, out),
398            IncomingMessage::Error(err) => self.fail_pending(&err.message),
399            // A refusal of a subscribe has an `errors` list in place of data. The parser of the session does not
400            // know that shape and hands it over as text.
401            IncomingMessage::Unknown(text) => {
402                if let Some(reason) = parse_subscribe_error(&text) {
403                    self.fail_pending(&reason);
404                }
405            }
406            IncomingMessage::Reconnected => self.on_reconnected(out),
407            _ => {}
408        }
409    }
410
411    fn poll_transmit(&mut self) -> Option<Outbound> {
412        self.outbound.pop_front()
413    }
414
415    fn poll_timeout(&self) -> Option<Instant> {
416        self.heartbeat_deadline
417    }
418
419    fn on_timeout(&mut self, _now: Instant, _out: &mut EventSink) {
420        self.heartbeat_deadline = None;
421    }
422
423    fn apply(&mut self, desired: &SubscriptionSet) {
424        let old_ids: Vec<_> = self.desired.subs().iter().map(|s| s.id).collect();
425        let new_ids: Vec<_> = desired.subs().iter().map(|s| s.id).collect();
426
427        let to_unsub: Vec<_> = self
428            .desired
429            .subs()
430            .iter()
431            .filter(|sub| !new_ids.contains(&sub.id))
432            .cloned()
433            .collect();
434        for sub in to_unsub {
435            self.queue_unsubscribe(&sub);
436        }
437
438        self.desired = desired.clone();
439        self.sub_by_id.clear();
440        for sub in self.desired.subs() {
441            self.sub_by_id.insert(sub.id, sub.clone());
442        }
443
444        let to_sub: Vec<_> = self
445            .desired
446            .subs()
447            .iter()
448            .filter(|sub| {
449                !old_ids.contains(&sub.id) || self.actual.get(&sub.id) == Some(&SubState::Failed)
450            })
451            .cloned()
452            .collect();
453        for sub in to_sub {
454            self.queue_subscribe(&sub);
455        }
456    }
457}
458
459fn bcs_key(instrument: &brokapi_core::InstrumentId) -> BcsInstrumentKey {
460    match &instrument.native {
461        brokapi_core::NativeId::TickerClass { ticker, class_code } => {
462            BcsInstrumentKey::new(ticker.clone(), class_code.clone())
463        }
464        _ => BcsInstrumentKey::new(instrument.key.symbol.clone(), instrument.key.mic.clone()),
465    }
466}
467
468fn same_instrument(
469    instrument: &brokapi_core::InstrumentId,
470    ticker: &str,
471    class_code: &str,
472) -> bool {
473    match &instrument.native {
474        brokapi_core::NativeId::TickerClass {
475            ticker: t,
476            class_code: c,
477        } => t == ticker && c == class_code,
478        _ => instrument.key.symbol == ticker && instrument.key.mic == class_code,
479    }
480}
481
482fn timeframe_label(tf: Timeframe) -> &'static str {
483    match tf {
484        Timeframe::M1 => "M1",
485        Timeframe::M5 => "M5",
486        // BCS has no M10: the connector refuses it before a subscription reaches the core.
487        Timeframe::M10 => "M10",
488        Timeframe::M15 => "M15",
489        Timeframe::M30 => "M30",
490        Timeframe::H1 => "H1",
491        Timeframe::H4 => "H4",
492        Timeframe::D1 => "D",
493        Timeframe::W1 => "W",
494        Timeframe::MN => "MN",
495    }
496}
497
498/// Detects a sequence gap and emits a resync: a hook for order book integrity.
499pub fn check_book_seq(
500    core: &mut MdCore,
501    sub: SubscriptionId,
502    expected: u64,
503    instrument: brokapi_core::InstrumentId,
504    out: &mut EventSink,
505) {
506    let seq = core.book_seq.get(&sub).copied().unwrap_or(0);
507    if seq > 0 && seq != expected {
508        out.push(CoreEvent {
509            sub: Some(sub),
510            payload: brokapi_core::Event::Resync {
511                sub,
512                instrument,
513                reason: ResyncReason::SequenceGap,
514            },
515        });
516    }
517}
518
519#[cfg(test)]
520mod tests {
521    use super::*;
522    use brokapi_core::{InstrumentId, InstrumentKey, NativeId};
523
524    fn sber() -> InstrumentId {
525        InstrumentId::new(
526            InstrumentKey::new("SBER", "TQBR"),
527            NativeId::TickerClass {
528                ticker: "SBER".into(),
529                class_code: "TQBR".into(),
530            },
531        )
532    }
533
534    #[test]
535    fn apply_queues_subscribe_outbound() {
536        let mut core = MdCore::new();
537        let sub_id = SubscriptionId::new(1);
538        let desired = SubscriptionSet::with_subscriptions(vec![DesiredSubscription {
539            id: sub_id,
540            kind: SubscriptionKind::OrderBook {
541                instrument: sber(),
542                depth: 20,
543            },
544        }]);
545        core.apply(&desired);
546        let msg = core.poll_transmit().expect("subscribe outbound");
547        match msg {
548            Outbound::WsText(json) => {
549                assert!(json.contains("\"subscribeType\":0"));
550                assert!(json.contains("\"dataType\":0"));
551            }
552        }
553        assert_eq!(core.actual_state(sub_id), Some(SubState::Pending));
554    }
555
556    #[test]
557    fn reconnect_marks_every_order_book_but_the_first_connect_does_not() {
558        let mut core = MdCore::new();
559        let (book, quotes) = (SubscriptionId::new(1), SubscriptionId::new(2));
560        core.apply(&SubscriptionSet::with_subscriptions(vec![
561            DesiredSubscription {
562                id: book,
563                kind: SubscriptionKind::OrderBook {
564                    instrument: sber(),
565                    depth: 20,
566                },
567            },
568            DesiredSubscription {
569                id: quotes,
570                kind: SubscriptionKind::Quotes { instrument: sber() },
571            },
572        ]));
573        let mut out = EventSink::new();
574        let now = Instant::now();
575
576        core.on_frame(IncomingMessage::Reconnected, now, &mut out);
577        assert!(out.drain().is_empty(), "the first connect is not a loss");
578
579        core.on_frame(IncomingMessage::Reconnected, now, &mut out);
580        let events = out.drain();
581        assert_eq!(
582            events.len(),
583            1,
584            "one Resync per order book subscription, none for quotes"
585        );
586        assert_eq!(events[0].sub, Some(book));
587        assert!(matches!(
588            events[0].payload,
589            Event::Resync { sub, reason: ResyncReason::SubscriptionReset, .. } if sub == book
590        ));
591        // The restored subscriptions are acknowledged again.
592        assert_eq!(core.actual_state(book), Some(SubState::Pending));
593        assert_eq!(core.actual_state(quotes), Some(SubState::Pending));
594    }
595}
brokapi_bcs/core/md.rs

brokapi_bcs/core/
md.rs
