brokapi_alor/core/
md.rs

1//! The market data core: frames of the data channel to events.
2
3use std::collections::{HashMap, HashSet, VecDeque};
4use std::sync::Arc;
5use std::time::Instant;
6
7use brokapi_alor_protocol::http::models::{
8    Bar, MarketTradeRecord, OrderBook as WireBook, QuoteRecord,
9};
10use brokapi_alor_protocol::ws::models::WsIncoming;
11use brokapi_core::{Event, InstrumentId, ResyncReason, SubscriptionId};
12use brokapi_sansio::{
13    BrokerCore, CoreEvent, EventSink, Outbound, SubscriptionKind, SubscriptionSet,
14};
15use tokio::sync::Notify;
16use tracing::{debug, warn};
17
18use crate::map::market;
19
20/// Where a subscription is in its life.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum SubState {
23    /// The request was sent, the broker has not confirmed it.
24    Pending,
25    /// The broker confirmed it, or data came.
26    Active,
27    /// The broker refused it.
28    Failed,
29}
30
31#[derive(Debug)]
32struct Entry {
33    id: SubscriptionId,
34    kind: SubscriptionKind,
35    state: SubState,
36    /// Why the broker refused the subscription.
37    reason: Option<String>,
38}
39
40/// How many trade ids a trades subscription remembers to drop a repeat.
41const SEEN_TRADES_CAP: usize = 20_000;
42
43/// The sans-io core of the market data connector.
44#[derive(Debug, Default)]
45pub struct MdCore {
46    desired: SubscriptionSet,
47    by_guid: HashMap<String, Entry>,
48    seen_trades: HashMap<SubscriptionId, (HashSet<i64>, VecDeque<i64>)>,
49    connections: u32,
50    ack: Arc<Notify>,
51}
52
53impl MdCore {
54    /// An empty core.
55    pub fn new() -> Self {
56        Self::default()
57    }
58
59    /// Woken whenever a subscription changes state (confirmed, refused, or data came). The connector waits on it
60    /// for the confirmation of a new subscription.
61    pub fn notifier(&self) -> Arc<Notify> {
62        Arc::clone(&self.ack)
63    }
64
65    /// The subscriptions the core was told about, as a set.
66    pub fn desired(&self) -> &SubscriptionSet {
67        &self.desired
68    }
69
70    /// Registers a subscription: frames with this `guid` belong to `id` and are read as `kind`. The connector
71    /// calls it before the request is sent, so the confirmation cannot overtake it.
72    pub fn register(&mut self, guid: &str, id: SubscriptionId, kind: SubscriptionKind) {
73        self.by_guid.insert(
74            guid.to_string(),
75            Entry {
76                id,
77                kind,
78                state: SubState::Pending,
79                reason: None,
80            },
81        );
82        self.refresh_desired();
83    }
84
85    /// Forgets a subscription (the data that still arrives for its `guid` is dropped).
86    pub fn forget(&mut self, guid: &str) {
87        if let Some(e) = self.by_guid.remove(guid) {
88            self.seen_trades.remove(&e.id);
89        }
90        self.refresh_desired();
91    }
92
93    fn refresh_desired(&mut self) {
94        let mut subs: Vec<_> = self
95            .by_guid
96            .values()
97            .map(|e| brokapi_sansio::DesiredSubscription {
98                id: e.id,
99                kind: e.kind.clone(),
100            })
101            .collect();
102        subs.sort_by_key(|s| s.id.value());
103        self.desired = SubscriptionSet::with_subscriptions(subs);
104    }
105
106    /// The state of a subscription.
107    pub fn actual_state(&self, id: SubscriptionId) -> Option<SubState> {
108        self.by_guid.values().find(|e| e.id == id).map(|e| e.state)
109    }
110
111    /// The reason the broker gave when it refused a subscription.
112    pub fn failure(&self, id: SubscriptionId) -> Option<&str> {
113        self.by_guid
114            .values()
115            .find(|e| e.id == id)
116            .and_then(|e| e.reason.as_deref())
117    }
118
119    fn on_reconnected(&mut self, out: &mut EventSink) {
120        self.connections += 1;
121        if self.connections < 2 {
122            return;
123        }
124        for e in self.by_guid.values_mut() {
125            e.state = SubState::Pending;
126            if let SubscriptionKind::OrderBook { instrument, .. } = &e.kind {
127                out.push(CoreEvent {
128                    sub: Some(e.id),
129                    payload: Event::Resync {
130                        sub: e.id,
131                        instrument: instrument.clone(),
132                        reason: ResyncReason::SubscriptionReset,
133                    },
134                });
135            }
136        }
137    }
138
139    fn on_data(&mut self, guid: &str, data: serde_json::Value, now: Instant, out: &mut EventSink) {
140        let Some(entry) = self.by_guid.get_mut(guid) else {
141            debug!("a frame of an unknown subscription {guid} is dropped");
142            return;
143        };
144        if entry.state != SubState::Active {
145            entry.state = SubState::Active;
146            self.ack.notify_waiters();
147        }
148        let id = entry.id;
149        let kind = entry.kind.clone();
150        let payload = match &kind {
151            SubscriptionKind::OrderBook { instrument, .. } => parse::<WireBook>(&data)
152                .map(|b| Event::OrderBook(market::order_book(&b, instrument, now))),
153            SubscriptionKind::Quotes { instrument } => parse::<QuoteRecord>(&data)
154                .map(|q| Event::Quote(market::quote(&q, instrument, now))),
155            SubscriptionKind::LastCandle {
156                instrument,
157                timeframe,
158            } => parse::<Bar>(&data)
159                .map(|b| Event::Candle(market::candle(&b, instrument, *timeframe, now))),
160            SubscriptionKind::LastTrades { instrument } => {
161                self.trade_event(id, instrument, &data, now)
162            }
163            _ => None,
164        };
165        if let Some(payload) = payload {
166            out.push(CoreEvent {
167                sub: Some(id),
168                payload,
169            });
170        }
171    }
172
173    fn trade_event(
174        &mut self,
175        id: SubscriptionId,
176        instrument: &InstrumentId,
177        data: &serde_json::Value,
178        now: Instant,
179    ) -> Option<Event> {
180        let rec = parse::<MarketTradeRecord>(data)?;
181        // The frames of the history that come with a subscription are not new trades.
182        if rec.existing == Some(true) {
183            return None;
184        }
185        if let Some(tid) = rec.id {
186            let (seen, order) = self.seen_trades.entry(id).or_default();
187            if !seen.insert(tid) {
188                return None;
189            }
190            order.push_back(tid);
191            if order.len() > SEEN_TRADES_CAP
192                && let Some(old) = order.pop_front()
193            {
194                seen.remove(&old);
195            }
196        }
197        market::market_trade(&rec, instrument, now).map(Event::MarketTrade)
198    }
199}
200
201fn parse<T: serde::de::DeserializeOwned>(data: &serde_json::Value) -> Option<T> {
202    match serde_json::from_value(data.clone()) {
203        Ok(v) => Some(v),
204        Err(e) => {
205            warn!("a data frame cannot be read: {e}");
206            None
207        }
208    }
209}
210
211impl BrokerCore for MdCore {
212    type Frame = WsIncoming;
213
214    fn on_frame(&mut self, frame: Self::Frame, now: Instant, out: &mut EventSink) {
215        match frame {
216            WsIncoming::Reconnected => self.on_reconnected(out),
217            WsIncoming::Data { guid, data } => self.on_data(&guid, data, now, out),
218            WsIncoming::Status {
219                guid: Some(guid),
220                code,
221                message,
222            } => {
223                if let Some(e) = self.by_guid.get_mut(&guid) {
224                    if code == 200 {
225                        e.state = SubState::Active;
226                    } else {
227                        e.state = SubState::Failed;
228                        e.reason = Some(message);
229                    }
230                    self.ack.notify_waiters();
231                }
232            }
233            WsIncoming::Status { .. } | WsIncoming::Unknown(_) => {}
234        }
235    }
236
237    fn poll_transmit(&mut self) -> Option<Outbound> {
238        // The requests go out through the WebSocket session of the connector, which also puts the token in.
239        None
240    }
241
242    fn poll_timeout(&self) -> Option<Instant> {
243        None
244    }
245
246    fn on_timeout(&mut self, _now: Instant, _out: &mut EventSink) {}
247
248    fn apply(&mut self, desired: &SubscriptionSet) {
249        self.desired = desired.clone();
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256    use crate::util::instrument_id;
257    use brokapi_core::Timeframe;
258    use serde_json::json;
259
260    fn sber() -> InstrumentId {
261        instrument_id("SBER", "TQBR")
262    }
263
264    fn data(guid: &str, d: serde_json::Value) -> WsIncoming {
265        WsIncoming::Data {
266            guid: guid.into(),
267            data: d,
268        }
269    }
270
271    #[test]
272    fn data_of_a_registered_guid_becomes_an_event_of_its_subscription() {
273        let mut core = MdCore::new();
274        let id = SubscriptionId::new(1);
275        core.register(
276            "g",
277            id,
278            SubscriptionKind::OrderBook {
279                instrument: sber(),
280                depth: 5,
281            },
282        );
283        let mut out = EventSink::new();
284        let now = Instant::now();
285        core.on_frame(
286            WsIncoming::Status {
287                guid: Some("g".into()),
288                code: 200,
289                message: "ok".into(),
290            },
291            now,
292            &mut out,
293        );
294        assert_eq!(core.actual_state(id), Some(SubState::Active));
295        core.on_frame(
296            data(
297                "g",
298                json!({"bids":[{"price":1.5,"volume":2}],"asks":[],"ms_timestamp":1}),
299            ),
300            now,
301            &mut out,
302        );
303        core.on_frame(data("other", json!({})), now, &mut out);
304        let events = out.drain();
305        assert_eq!(events.len(), 1);
306        assert_eq!(events[0].sub, Some(id));
307        assert!(matches!(events[0].payload, Event::OrderBook(_)));
308    }
309
310    #[test]
311    fn a_refused_subscription_is_failed_with_the_reason() {
312        let mut core = MdCore::new();
313        let id = SubscriptionId::new(1);
314        core.register("g", id, SubscriptionKind::Quotes { instrument: sber() });
315        let mut out = EventSink::new();
316        core.on_frame(
317            WsIncoming::Status {
318                guid: Some("g".into()),
319                code: 400,
320                message: "bad symbol".into(),
321            },
322            Instant::now(),
323            &mut out,
324        );
325        assert_eq!(core.actual_state(id), Some(SubState::Failed));
326        assert_eq!(core.failure(id), Some("bad symbol"));
327    }
328
329    #[test]
330    fn trades_of_the_history_and_repeats_are_dropped() {
331        let mut core = MdCore::new();
332        core.register(
333            "t",
334            SubscriptionId::new(2),
335            SubscriptionKind::LastTrades { instrument: sber() },
336        );
337        let mut out = EventSink::new();
338        let now = Instant::now();
339        let t = |id: i64, existing: bool| {
340            data(
341                "t",
342                json!({"id": id, "qty": 1, "price": 10, "side": "buy", "existing": existing}),
343            )
344        };
345        core.on_frame(t(1, true), now, &mut out);
346        core.on_frame(t(2, false), now, &mut out);
347        core.on_frame(t(2, false), now, &mut out);
348        core.on_frame(t(3, false), now, &mut out);
349        assert_eq!(out.drain().len(), 2);
350    }
351
352    #[test]
353    fn a_reconnect_voids_the_books_but_the_first_connect_does_not() {
354        let mut core = MdCore::new();
355        let (book, quotes) = (SubscriptionId::new(1), SubscriptionId::new(2));
356        core.register(
357            "b",
358            book,
359            SubscriptionKind::OrderBook {
360                instrument: sber(),
361                depth: 20,
362            },
363        );
364        core.register("q", quotes, SubscriptionKind::Quotes { instrument: sber() });
365        let mut out = EventSink::new();
366        let now = Instant::now();
367        core.on_frame(WsIncoming::Reconnected, now, &mut out);
368        assert!(out.drain().is_empty(), "the first connect is not a loss");
369        core.on_frame(WsIncoming::Reconnected, now, &mut out);
370        let events = out.drain();
371        assert_eq!(
372            events.len(),
373            1,
374            "one Resync per order book, none for quotes"
375        );
376        assert!(matches!(
377            events[0].payload,
378            Event::Resync { sub, reason: ResyncReason::SubscriptionReset, .. } if sub == book
379        ));
380        assert_eq!(core.actual_state(quotes), Some(SubState::Pending));
381    }
382
383    #[test]
384    fn a_candle_frame_is_the_candle_of_the_timeframe() {
385        let mut core = MdCore::new();
386        core.register(
387            "c",
388            SubscriptionId::new(3),
389            SubscriptionKind::LastCandle {
390                instrument: sber(),
391                timeframe: Timeframe::M5,
392            },
393        );
394        let mut out = EventSink::new();
395        core.on_frame(
396            data("c", json!({"time": 1_700_000_000, "open": 1, "high": 2, "low": 1, "close": 2, "volume": 5})),
397            Instant::now(),
398            &mut out,
399        );
400        let events = out.drain();
401        let Event::Candle(c) = &events[0].payload else {
402            panic!("a candle")
403        };
404        assert_eq!(c.timeframe, Timeframe::M5);
405    }
406
407    #[test]
408    fn a_refused_subscription_keeps_the_reason_and_a_bad_or_repeated_frame_gives_nothing() {
409        let mut core = MdCore::new();
410        let id = SubscriptionId::new(1);
411        core.register("g", id, SubscriptionKind::LastTrades { instrument: sber() });
412        assert_eq!(core.desired().subs().len(), 1);
413        let mut out = EventSink::new();
414        let now = Instant::now();
415        core.on_frame(
416            WsIncoming::Status {
417                guid: Some("g".into()),
418                code: 400,
419                message: "no such instrument".into(),
420            },
421            now,
422            &mut out,
423        );
424        assert_eq!(core.actual_state(id), Some(SubState::Failed));
425        assert_eq!(core.failure(id), Some("no such instrument"));
426        // A status without a guid and a frame nobody knows are not for any subscription.
427        core.on_frame(
428            WsIncoming::Status {
429                guid: None,
430                code: 200,
431                message: String::new(),
432            },
433            now,
434            &mut out,
435        );
436        core.on_frame(WsIncoming::Unknown("x".into()), now, &mut out);
437        // A trade that cannot be read is dropped, the same trade twice is told once.
438        core.on_frame(data("g", json!({"id": "not a number"})), now, &mut out);
439        let trade = json!({"id": 7, "price": 100.5, "qty": 3, "timestamp": 1, "side": "buy", "symbol": "SBER"});
440        core.on_frame(data("g", trade.clone()), now, &mut out);
441        core.on_frame(data("g", trade), now, &mut out);
442        assert_eq!(out.drain().len(), 1);
443    }
444
445    #[test]
446    fn a_trades_subscription_forgets_the_oldest_ids_when_the_memory_is_full() {
447        let mut core = MdCore::new();
448        let id = SubscriptionId::new(1);
449        core.register("g", id, SubscriptionKind::LastTrades { instrument: sber() });
450        let mut out = EventSink::new();
451        let now = Instant::now();
452        let trade = |n: i64| json!({"id": n, "price": 100.5, "qty": 3, "timestamp": 1, "side": "buy", "symbol": "SBER"});
453        for n in 0..=SEEN_TRADES_CAP as i64 {
454            core.on_frame(data("g", trade(n)), now, &mut out);
455        }
456        assert_eq!(out.drain().len(), SEEN_TRADES_CAP + 1);
457        // Trade 0 fell out of the memory: it is told again, trade 1 is still a repeat.
458        core.on_frame(data("g", trade(0)), now, &mut out);
459        core.on_frame(data("g", trade(SEEN_TRADES_CAP as i64)), now, &mut out);
460        assert_eq!(out.drain().len(), 1);
461    }
462
463    #[test]
464    fn the_core_sends_nothing_by_itself_and_has_no_timers() {
465        let mut core = MdCore::new();
466        assert!(core.poll_transmit().is_none());
467        assert!(core.poll_timeout().is_none());
468        core.on_timeout(Instant::now(), &mut EventSink::new());
469        let mut set = SubscriptionSet::new();
470        core.apply(&set);
471        set = SubscriptionSet::with_subscriptions(vec![]);
472        core.apply(&set);
473        assert!(core.desired().is_empty());
474    }
475}
brokapi_alor/core/md.rs

brokapi_alor/core/
md.rs
