brokapi_finam/core/
md.rs

1//! Market-data core: books rebuilt from action rows, quotes, trades, bars.
2
3use std::collections::{HashMap, HashSet, VecDeque};
4use std::time::Instant;
5
6use brokapi_core::{Event, InstrumentId, ResyncReason, SubscriptionId, Timeframe};
7use brokapi_finam_protocol::pb::grpc::tradeapi::v1::marketdata as pb;
8use brokapi_sansio::{
9    BrokerCore, CoreEvent, DesiredSubscription, EventSink, Outbound, SubscriptionKind,
10    SubscriptionSet,
11};
12
13use super::{StreamEvent, SubStatus};
14use crate::map::m10::M10Builder;
15use crate::map::market::{
16    ApplyOutcome, BookState, book_snapshot, map_bar, map_quote, map_trade, stream_rows,
17};
18
19/// Default depth of emitted books when the subscription did not ask for one.
20pub const DEFAULT_DEPTH: u32 = 50;
21
22/// What arrived on a market-data stream.
23#[derive(Debug, Clone)]
24pub enum MdEvent {
25    /// A message of an order book stream: snapshot or action rows.
26    OrderBook(pb::SubscribeOrderBookResponse),
27    /// A message of a quote stream. It can carry an in-band error instead of quotes.
28    Quotes(pb::SubscribeQuoteResponse),
29    /// A message of a market trades stream.
30    Trades(pb::SubscribeLatestTradesResponse),
31    /// A message of a bars stream.
32    Bars(pb::SubscribeBarsResponse),
33    /// The stream was opened (also after a reopen).
34    Connected,
35    /// The stream broke or ended and is being reopened. The text says why.
36    Disconnected(String),
37    /// The broker refuses the stream (bad symbol, no permission). The text says why.
38    Failed(String),
39}
40
41/// One stream event of one subscription.
42#[derive(Debug, Clone)]
43pub struct MdFrame {
44    /// The subscription whose stream produced the event.
45    pub sub: SubscriptionId,
46    /// What happened on that stream.
47    pub event: MdEvent,
48}
49
50impl MdFrame {
51    /// Frame for a stream lifecycle event (connected, disconnected, failed).
52    ///
53    /// `None` for [`StreamEvent::Msg`], because a message has its own typed variant of [`MdEvent`].
54    ///
55    /// # Examples
56    ///
57    /// ```
58    /// use brokapi_core::SubscriptionId;
59    /// use brokapi_finam::core::{MdEvent, MdFrame, StreamEvent};
60    ///
61    /// let sub = SubscriptionId::new(1);
62    /// let frame = MdFrame::lifecycle(sub, StreamEvent::Connected).unwrap();
63    /// assert!(matches!(frame.event, MdEvent::Connected));
64    /// assert!(MdFrame::lifecycle(sub, StreamEvent::Msg(())).is_none());
65    /// ```
66    pub fn lifecycle(sub: SubscriptionId, ev: StreamEvent<()>) -> Option<Self> {
67        let event = match ev {
68            StreamEvent::Connected => MdEvent::Connected,
69            StreamEvent::Disconnected(r) => MdEvent::Disconnected(r),
70            StreamEvent::Failed(r) => MdEvent::Failed(r),
71            StreamEvent::Msg(()) => return None,
72        };
73        Some(Self { sub, event })
74    }
75}
76
77#[derive(Debug, Default)]
78struct BookEntry {
79    state: BookState,
80    /// How many times the stream has been (re)opened.
81    connects: u32,
82}
83
84#[derive(Debug, Default)]
85struct SeenTrades {
86    ids: HashSet<String>,
87    order: VecDeque<String>,
88}
89
90impl SeenTrades {
91    const CAP: usize = 2048;
92
93    /// `true` the first time an id is seen. Streams replay recent trades after a reconnect.
94    fn first_time(&mut self, id: &str) -> bool {
95        if id.is_empty() {
96            return true;
97        }
98        if !self.ids.insert(id.to_string()) {
99            return false;
100        }
101        self.order.push_back(id.to_string());
102        if self.order.len() > Self::CAP
103            && let Some(old) = self.order.pop_front()
104        {
105            self.ids.remove(&old);
106        }
107        true
108    }
109}
110
111/// Market data core: turns stream frames into events and keeps the state behind them.
112///
113/// Per subscription it holds the local order book (rebuilt from action rows), the ids of trades
114/// already seen (a reconnect replays recent trades, a repeat is dropped), the M10 candle being built
115/// from M5 bars, and the stream status.
116///
117/// Implements `BrokerCore` from `brokapi-sansio`. See the [module docs](crate::core) for an example.
118#[derive(Debug, Default)]
119pub struct MdCore {
120    desired: SubscriptionSet,
121    status: HashMap<SubscriptionId, SubStatus>,
122    books: HashMap<SubscriptionId, BookEntry>,
123    seen_trades: HashMap<SubscriptionId, SeenTrades>,
124    /// M10 subscriptions: the stream is M5, the 10-minute candle is built here.
125    m10: HashMap<SubscriptionId, M10Builder>,
126    /// Subscriptions whose stream must be reopened (book out of sync).
127    restarts: Vec<SubscriptionId>,
128}
129
130impl MdCore {
131    /// An empty core with no subscriptions.
132    pub fn new() -> Self {
133        Self::default()
134    }
135
136    /// The subscriptions the core was told to serve (set by `apply`).
137    pub fn desired(&self) -> &SubscriptionSet {
138        &self.desired
139    }
140
141    /// Stream status of subscription `sub`. `None` for an id the core does not know.
142    pub fn status(&self, sub: SubscriptionId) -> Option<SubStatus> {
143        self.status.get(&sub).cloned()
144    }
145
146    /// Streams to reopen because their local book became unusable. Drains the list.
147    pub fn take_restarts(&mut self) -> Vec<SubscriptionId> {
148        std::mem::take(&mut self.restarts)
149    }
150
151    fn kind(&self, sub: SubscriptionId) -> Option<&SubscriptionKind> {
152        self.desired.find(sub).map(|d| &d.kind)
153    }
154
155    fn on_order_book(
156        &mut self,
157        sub: SubscriptionId,
158        instrument: InstrumentId,
159        depth: u32,
160        msg: pb::StreamOrderBook,
161        now: Instant,
162        out: &mut EventSink,
163    ) {
164        let rows = stream_rows(&msg.rows);
165        let entry = self.books.entry(sub).or_default();
166        match entry.state.apply(&rows, msg.is_data_snapshot) {
167            ApplyOutcome::NeedSnapshot => {
168                // Increment without a base: drop it and ask for a fresh stream (which starts
169                // with a snapshot). The consumer is told the book is not trustworthy.
170                self.restarts.push(sub);
171                out.push(CoreEvent {
172                    sub: Some(sub),
173                    payload: Event::Resync {
174                        sub,
175                        instrument,
176                        reason: ResyncReason::SequenceGap,
177                    },
178                });
179            }
180            ApplyOutcome::Applied => {
181                let broker_ts = rows.iter().filter_map(|r| r.ts).max();
182                let book = book_snapshot(&instrument, &entry.state, depth as usize, broker_ts, now);
183                out.push(CoreEvent {
184                    sub: Some(sub),
185                    payload: Event::OrderBook(book),
186                });
187            }
188        }
189    }
190}
191
192impl BrokerCore for MdCore {
193    type Frame = MdFrame;
194
195    fn on_frame(&mut self, frame: MdFrame, now: Instant, out: &mut EventSink) {
196        let MdFrame { sub, event } = frame;
197        let Some(kind) = self.kind(sub).cloned() else {
198            // Late frame of an unsubscribed stream.
199            return;
200        };
201        match event {
202            MdEvent::Connected => {
203                self.status.insert(sub, SubStatus::Active);
204                if let SubscriptionKind::OrderBook { instrument, .. } = &kind {
205                    // A (re)opened stream starts with a snapshot: the old book is void.
206                    let entry = self.books.entry(sub).or_default();
207                    let reopened = entry.connects > 0;
208                    entry.state.reset();
209                    entry.connects += 1;
210                    if reopened {
211                        out.push(CoreEvent {
212                            sub: Some(sub),
213                            payload: Event::Resync {
214                                sub,
215                                instrument: instrument.clone(),
216                                reason: ResyncReason::SubscriptionReset,
217                            },
218                        });
219                    }
220                }
221            }
222            MdEvent::Disconnected(r) => {
223                self.status.insert(sub, SubStatus::Reconnecting(r));
224            }
225            MdEvent::Failed(r) => {
226                self.status.insert(sub, SubStatus::Failed(r));
227            }
228            MdEvent::OrderBook(msg) => {
229                if let SubscriptionKind::OrderBook { instrument, depth } = kind {
230                    self.status.insert(sub, SubStatus::Active);
231                    let symbol = format!("{}@{}", instrument.key.symbol, instrument.key.mic);
232                    for book in msg.order_book {
233                        if !book.symbol.is_empty() && book.symbol != symbol {
234                            continue;
235                        }
236                        self.on_order_book(sub, instrument.clone(), depth, book, now, out);
237                    }
238                }
239            }
240            MdEvent::Quotes(msg) => {
241                if let SubscriptionKind::Quotes { instrument } = kind {
242                    if let Some(e) = &msg.error {
243                        self.status.insert(
244                            sub,
245                            SubStatus::Erroring(format!("{}: {}", e.code, e.description)),
246                        );
247                        return;
248                    }
249                    self.status.insert(sub, SubStatus::Active);
250                    let symbol = format!("{}@{}", instrument.key.symbol, instrument.key.mic);
251                    for q in &msg.quote {
252                        // One stream per subscription, but be strict about the symbol.
253                        if !q.symbol.is_empty() && q.symbol != symbol {
254                            continue;
255                        }
256                        out.push(CoreEvent {
257                            sub: Some(sub),
258                            payload: Event::Quote(map_quote(&instrument, q, now)),
259                        });
260                    }
261                }
262            }
263            MdEvent::Trades(msg) => {
264                if let SubscriptionKind::LastTrades { instrument } = kind {
265                    self.status.insert(sub, SubStatus::Active);
266                    let seen = self.seen_trades.entry(sub).or_default();
267                    for t in &msg.trades {
268                        if !seen.first_time(&t.trade_id) {
269                            continue;
270                        }
271                        if let Some(trade) = map_trade(&instrument, t, now) {
272                            out.push(CoreEvent {
273                                sub: Some(sub),
274                                payload: Event::MarketTrade(trade),
275                            });
276                        }
277                    }
278                }
279            }
280            MdEvent::Bars(msg) => {
281                if let SubscriptionKind::LastCandle {
282                    instrument,
283                    timeframe,
284                } = kind
285                {
286                    self.status.insert(sub, SubStatus::Active);
287                    for b in &msg.bars {
288                        let candle = if timeframe == Timeframe::M10 {
289                            // The wire stream is M5.
290                            let m5 = map_bar(&instrument, Timeframe::M5, b, now);
291                            match self.m10.entry(sub).or_default().on_m5(&m5) {
292                                Some(c) => c,
293                                None => continue,
294                            }
295                        } else {
296                            map_bar(&instrument, timeframe, b, now)
297                        };
298                        out.push(CoreEvent {
299                            sub: Some(sub),
300                            payload: Event::Candle(candle),
301                        });
302                    }
303                }
304            }
305        }
306    }
307
308    /// gRPC streams are opened by the stream tasks, nothing is queued for a socket.
309    fn poll_transmit(&mut self) -> Option<Outbound> {
310        None
311    }
312
313    fn poll_timeout(&self) -> Option<Instant> {
314        None
315    }
316
317    fn on_timeout(&mut self, _now: Instant, _out: &mut EventSink) {}
318
319    fn apply(&mut self, desired: &SubscriptionSet) {
320        let keep: HashSet<SubscriptionId> = desired.iter().map(|d| d.id).collect();
321        self.status.retain(|id, _| keep.contains(id));
322        self.books.retain(|id, _| keep.contains(id));
323        self.seen_trades.retain(|id, _| keep.contains(id));
324        self.m10.retain(|id, _| keep.contains(id));
325        for DesiredSubscription { id, kind } in desired.iter() {
326            self.status.entry(*id).or_insert(SubStatus::Connecting);
327            if matches!(kind, SubscriptionKind::OrderBook { .. }) {
328                self.books.entry(*id).or_default();
329            }
330        }
331        self.desired = desired.clone();
332    }
333}
brokapi_finam/core/md.rs

brokapi_finam/core/
md.rs
