brokapi_alor/core/
ud.rs

1//! The user data core: frames of the data channel (own orders, fills, positions) to events.
2
3use std::collections::{HashMap, HashSet, VecDeque};
4use std::sync::{Arc, Mutex};
5use std::time::Instant;
6
7use brokapi_alor_protocol::http::models::{OrderRecord, PositionRecord, TradeRecord};
8use brokapi_alor_protocol::ws::models::WsIncoming;
9use brokapi_core::{ClientOrderId, Event, SubscriptionId};
10use brokapi_sansio::{
11    BrokerCore, CoreEvent, DesiredSubscription, EventSink, Outbound, SubscriptionKind,
12    SubscriptionSet,
13};
14use tokio::sync::Notify;
15use tracing::warn;
16
17use crate::cid::{cid_from_comment, foreign_cid};
18use crate::core::md::SubState;
19use crate::map::{execution, order, portfolio};
20use crate::subs::UdSubKind;
21
22/// The boards of the instruments of a portfolio, by `EXCHANGE:TICKER`. The positions of the broker do not name
23/// the board of an instrument, so the connector looks it up and keeps it here, where the core reads it.
24pub type BoardBook = Arc<Mutex<HashMap<String, String>>>;
25
26/// The key of an instrument in a [`BoardBook`].
27pub fn board_key(exchange: &str, ticker: &str) -> String {
28    format!("{exchange}:{ticker}")
29}
30
31const SEEN_EXECUTIONS_CAP: usize = 20_000;
32const ORDER_IDS_CAP: usize = 20_000;
33
34/// The sans-io core of the user data connector.
35#[derive(Debug)]
36pub struct UdCore {
37    desired: SubscriptionSet,
38    by_guid: HashMap<String, (SubscriptionId, UdSubKind)>,
39    seen_executions: HashSet<String>,
40    seen_order: VecDeque<String>,
41    order_ids: HashMap<String, ClientOrderId>,
42    order_ids_order: VecDeque<String>,
43    boards: BoardBook,
44    exchange: String,
45    connects: u32,
46    states: HashMap<SubscriptionId, (SubState, Option<String>)>,
47    ack: Arc<Notify>,
48}
49
50impl UdCore {
51    /// An empty core for the exchange of the portfolio, with the book of boards it reads positions from.
52    pub fn new(exchange: impl Into<String>, boards: BoardBook) -> Self {
53        Self {
54            desired: SubscriptionSet::new(),
55            by_guid: HashMap::new(),
56            seen_executions: HashSet::new(),
57            seen_order: VecDeque::new(),
58            order_ids: HashMap::new(),
59            order_ids_order: VecDeque::new(),
60            boards,
61            exchange: exchange.into(),
62            connects: 0,
63            states: HashMap::new(),
64            ack: Arc::new(Notify::new()),
65        }
66    }
67
68    /// Woken whenever a subscription changes state. The connector waits on it for the confirmation of a new
69    /// subscription.
70    pub fn notifier(&self) -> Arc<Notify> {
71        Arc::clone(&self.ack)
72    }
73
74    /// The state of a subscription.
75    pub fn actual_state(&self, id: SubscriptionId) -> Option<SubState> {
76        self.states.get(&id).map(|(s, _)| *s)
77    }
78
79    /// The reason the broker gave when it refused a subscription.
80    pub fn failure(&self, id: SubscriptionId) -> Option<&str> {
81        self.states.get(&id).and_then(|(_, r)| r.as_deref())
82    }
83
84    /// Registers a subscription: frames with this `guid` belong to `id` and are read as `kind`.
85    pub fn register(&mut self, guid: &str, id: SubscriptionId, kind: UdSubKind) {
86        self.by_guid.insert(guid.to_string(), (id, kind));
87        self.states.insert(id, (SubState::Pending, None));
88        self.refresh_desired();
89    }
90
91    /// Forgets a subscription.
92    pub fn forget(&mut self, guid: &str) {
93        if let Some((id, _)) = self.by_guid.remove(guid) {
94            self.states.remove(&id);
95        }
96        self.refresh_desired();
97    }
98
99    fn refresh_desired(&mut self) {
100        let mut subs: Vec<DesiredSubscription> = self
101            .by_guid
102            .values()
103            .map(|(id, kind)| DesiredSubscription {
104                id: *id,
105                kind: match kind {
106                    UdSubKind::OrderState => SubscriptionKind::OrderState,
107                    UdSubKind::Portfolio => SubscriptionKind::Portfolio,
108                    UdSubKind::Executions => SubscriptionKind::Executions,
109                },
110            })
111            .collect();
112        subs.sort_by_key(|s| s.id.value());
113        self.desired = SubscriptionSet::with_subscriptions(subs);
114    }
115
116    /// Remembers that the order with this number is the order of `cid`.
117    pub fn learn_order_id(&mut self, order_id: &str, cid: ClientOrderId) {
118        if order_id.is_empty() || cid.is_nil() {
119            return;
120        }
121        if self.order_ids.insert(order_id.to_string(), cid).is_none() {
122            self.order_ids_order.push_back(order_id.to_string());
123            if self.order_ids_order.len() > ORDER_IDS_CAP
124                && let Some(old) = self.order_ids_order.pop_front()
125            {
126                self.order_ids.remove(&old);
127            }
128        }
129    }
130
131    /// The cid of the order with this number, if the core has seen it.
132    pub fn cid_of_order_id(&self, order_id: &str) -> Option<ClientOrderId> {
133        self.order_ids.get(order_id).copied()
134    }
135
136    /// The cid of an order of the broker: the one in its comment, else the one learned for its number, else a
137    /// stable made-up one (the order is not ours).
138    pub fn resolve_cid(&mut self, order_id: &str, comment: Option<&str>) -> ClientOrderId {
139        if let Some(cid) = cid_from_comment(comment) {
140            self.learn_order_id(order_id, cid);
141            return cid;
142        }
143        self.cid_of_order_id(order_id)
144            .unwrap_or_else(|| foreign_cid(order_id))
145    }
146
147    fn find(&self, kind: UdSubKind) -> Option<SubscriptionId> {
148        self.by_guid
149            .values()
150            .find(|(_, k)| *k == kind)
151            .map(|(id, _)| *id)
152    }
153
154    fn on_connect(&mut self, out: &mut EventSink) {
155        self.connects += 1;
156        if self.connects < 2 {
157            return;
158        }
159        for (state, _) in self.states.values_mut() {
160            *state = SubState::Pending;
161        }
162        let gap_from = crate::map::unix_nanos(chrono::Utc::now());
163        for kind in [UdSubKind::OrderState, UdSubKind::Executions] {
164            if let Some(sub) = self.find(kind) {
165                out.push(CoreEvent {
166                    sub: Some(sub),
167                    payload: Event::Reconnected { sub, gap_from },
168                });
169            }
170        }
171    }
172
173    fn on_data(&mut self, guid: &str, data: serde_json::Value, now: Instant, out: &mut EventSink) {
174        let Some(&(id, kind)) = self.by_guid.get(guid) else {
175            return;
176        };
177        if self
178            .states
179            .get(&id)
180            .is_some_and(|(s, _)| *s != SubState::Active)
181        {
182            self.states.insert(id, (SubState::Active, None));
183            self.ack.notify_waiters();
184        }
185        let payload = match kind {
186            UdSubKind::OrderState => read::<OrderRecord>(&data).and_then(|r| {
187                let cid = self.resolve_cid(&r.id, r.comment.as_deref());
188                order::order_snapshot(&r, cid, now).map(Event::Order)
189            }),
190            UdSubKind::Executions => read::<TradeRecord>(&data).and_then(|r| {
191                // A trade the history of the subscription repeats, or one seen twice, is told once.
192                if !self.seen_executions.insert(r.id.clone()) {
193                    return None;
194                }
195                self.seen_order.push_back(r.id.clone());
196                if self.seen_order.len() > SEEN_EXECUTIONS_CAP
197                    && let Some(old) = self.seen_order.pop_front()
198                {
199                    self.seen_executions.remove(&old);
200                }
201                let order_no = r.orderno.clone().unwrap_or_default();
202                let cid = cid_from_comment(r.comment.as_deref())
203                    .or_else(|| self.cid_of_order_id(&order_no));
204                execution::execution(&r, cid, now).map(Event::Execution)
205            }),
206            UdSubKind::Portfolio => read::<PositionRecord>(&data).map(|r| {
207                if portfolio::is_money(&r) {
208                    Event::Balances(vec![portfolio::balance(&r, now)])
209                } else {
210                    let board = self
211                        .boards
212                        .lock()
213                        .ok()
214                        .and_then(|b| b.get(&board_key(&self.exchange, &r.symbol)).cloned())
215                        .unwrap_or_default();
216                    Event::Positions(vec![portfolio::position(&r, &board, now)])
217                }
218            }),
219        };
220        if let Some(payload) = payload {
221            out.push(CoreEvent {
222                sub: Some(id),
223                payload,
224            });
225        }
226    }
227}
228
229fn read<T: serde::de::DeserializeOwned>(data: &serde_json::Value) -> Option<T> {
230    match serde_json::from_value(data.clone()) {
231        Ok(v) => Some(v),
232        Err(e) => {
233            warn!("a data frame cannot be read: {e}");
234            None
235        }
236    }
237}
238
239impl BrokerCore for UdCore {
240    type Frame = WsIncoming;
241
242    fn on_frame(&mut self, frame: Self::Frame, now: Instant, out: &mut EventSink) {
243        match frame {
244            WsIncoming::Reconnected => self.on_connect(out),
245            WsIncoming::Data { guid, data } => self.on_data(&guid, data, now, out),
246            WsIncoming::Status {
247                guid: Some(guid),
248                code,
249                message,
250            } => {
251                if let Some(&(id, _)) = self.by_guid.get(&guid) {
252                    let st = if code == 200 {
253                        (SubState::Active, None)
254                    } else {
255                        (SubState::Failed, Some(message))
256                    };
257                    self.states.insert(id, st);
258                    self.ack.notify_waiters();
259                }
260            }
261            WsIncoming::Status { .. } | WsIncoming::Unknown(_) => {}
262        }
263    }
264
265    fn poll_transmit(&mut self) -> Option<Outbound> {
266        None
267    }
268
269    fn poll_timeout(&self) -> Option<Instant> {
270        None
271    }
272
273    fn on_timeout(&mut self, _now: Instant, _out: &mut EventSink) {}
274
275    fn apply(&mut self, desired: &SubscriptionSet) {
276        self.desired = desired.clone();
277    }
278}
279
280#[cfg(test)]
281mod tests {
282    use super::*;
283    use serde_json::json;
284    use uuid::Uuid;
285
286    fn core() -> UdCore {
287        UdCore::new("MOEX", BoardBook::default())
288    }
289
290    fn data(guid: &str, d: serde_json::Value) -> WsIncoming {
291        WsIncoming::Data {
292            guid: guid.into(),
293            data: d,
294        }
295    }
296
297    #[test]
298    fn the_cid_comes_from_the_comment_and_is_learned_for_the_number() {
299        let mut c = core();
300        let id = SubscriptionId::new(1001);
301        c.register("o", id, UdSubKind::OrderState);
302        let cid = Uuid::new_v4();
303        let mut out = EventSink::new();
304        let now = Instant::now();
305        let rec = |status: &str, comment: Option<String>| {
306            json!({"id":"77","symbol":"SBER","board":"TQBR","type":"limit","side":"buy","status":status,
307                   "qtyUnits":20,"filledQtyUnits":0,"price":100,"comment":comment})
308        };
309        c.on_frame(
310            data("o", rec("working", Some(cid.to_string()))),
311            now,
312            &mut out,
313        );
314        // Later frames of the same order may come without the comment: the number is known.
315        c.on_frame(data("o", rec("canceled", None)), now, &mut out);
316        let events = out.drain();
317        assert_eq!(events.len(), 2);
318        for e in &events {
319            let Event::Order(s) = &e.payload else {
320                panic!("an order")
321            };
322            assert_eq!(s.cid, cid);
323            assert_eq!(e.sub, Some(id));
324        }
325    }
326
327    #[test]
328    fn an_order_that_is_not_ours_gets_a_stable_made_up_cid() {
329        let mut c = core();
330        assert_eq!(
331            c.resolve_cid("500", Some("someone else's note")),
332            foreign_cid("500")
333        );
334        assert_eq!(c.resolve_cid("500", None), foreign_cid("500"));
335    }
336
337    #[test]
338    fn a_fill_is_told_once_and_carries_the_cid_of_its_order() {
339        let mut c = core();
340        c.register("t", SubscriptionId::new(1002), UdSubKind::Executions);
341        let cid = Uuid::new_v4();
342        c.learn_order_id("77", cid);
343        let mut out = EventSink::new();
344        let f = || {
345            data(
346                "t",
347                json!({"id":"T1","orderno":"77","symbol":"SBER","board":"TQBR","qtyUnits":20,"price":100.5,"side":"buy"}),
348            )
349        };
350        c.on_frame(f(), Instant::now(), &mut out);
351        c.on_frame(f(), Instant::now(), &mut out);
352        let events = out.drain();
353        assert_eq!(events.len(), 1);
354        let Event::Execution(e) = &events[0].payload else {
355            panic!("an execution")
356        };
357        assert_eq!(e.cid, Some(cid));
358    }
359
360    #[test]
361    fn a_position_uses_the_board_the_connector_looked_up() {
362        let boards = BoardBook::default();
363        boards
364            .lock()
365            .unwrap()
366            .insert(board_key("MOEX", "SBER"), "TQBR".into());
367        let mut c = UdCore::new("MOEX", boards);
368        c.register("p", SubscriptionId::new(1003), UdSubKind::Portfolio);
369        let mut out = EventSink::new();
370        c.on_frame(
371            data(
372                "p",
373                json!({"symbol":"SBER","qtyUnits":30,"isCurrency":false}),
374            ),
375            Instant::now(),
376            &mut out,
377        );
378        c.on_frame(
379            data(
380                "p",
381                json!({"symbol":"RUB","qtyUnits":1000,"isCurrency":true}),
382            ),
383            Instant::now(),
384            &mut out,
385        );
386        let events = out.drain();
387        let Event::Positions(p) = &events[0].payload else {
388            panic!("positions")
389        };
390        assert_eq!(p[0].instrument.key.display(), "SBER@TQBR");
391        assert!(matches!(events[1].payload, Event::Balances(_)));
392    }
393
394    #[test]
395    fn a_reconnect_tells_the_order_and_fill_subscriptions_from_the_second_connect_on() {
396        let mut c = core();
397        c.register("o", SubscriptionId::new(1001), UdSubKind::OrderState);
398        c.register("p", SubscriptionId::new(1003), UdSubKind::Portfolio);
399        let mut out = EventSink::new();
400        c.on_frame(WsIncoming::Reconnected, Instant::now(), &mut out);
401        assert!(out.drain().is_empty());
402        c.on_frame(WsIncoming::Reconnected, Instant::now(), &mut out);
403        let events = out.drain();
404        assert_eq!(events.len(), 1);
405        assert!(matches!(events[0].payload, Event::Reconnected { .. }));
406    }
407
408    #[test]
409    fn a_refused_subscription_keeps_the_reason_and_frames_nobody_wants_give_nothing() {
410        let mut c = core();
411        let id = SubscriptionId::new(1001);
412        c.register("o", id, UdSubKind::OrderState);
413        let mut out = EventSink::new();
414        let now = Instant::now();
415        c.on_frame(
416            WsIncoming::Status {
417                guid: Some("o".into()),
418                code: 400,
419                message: "refused".into(),
420            },
421            now,
422            &mut out,
423        );
424        assert_eq!(c.actual_state(id), Some(SubState::Failed));
425        assert_eq!(c.failure(id), Some("refused"));
426        c.on_frame(
427            WsIncoming::Status {
428                guid: None,
429                code: 200,
430                message: String::new(),
431            },
432            now,
433            &mut out,
434        );
435        c.on_frame(WsIncoming::Unknown("x".into()), now, &mut out);
436        // Data of a guid nobody registered, and data that is not an order.
437        c.on_frame(data("other", json!({})), now, &mut out);
438        c.on_frame(data("o", json!({"id": 5})), now, &mut out);
439        assert!(out.drain().is_empty());
440        // Forgotten, the subscription has no state.
441        c.forget("o");
442        assert_eq!(c.actual_state(id), None);
443    }
444
445    #[test]
446    fn the_memory_of_order_numbers_and_trades_is_bounded() {
447        let mut c = core();
448        let cid = Uuid::new_v4();
449        c.learn_order_id("", cid);
450        c.learn_order_id("1", Uuid::nil());
451        assert_eq!(c.cid_of_order_id("1"), None);
452        for n in 0..=ORDER_IDS_CAP {
453            c.learn_order_id(&n.to_string(), cid);
454        }
455        assert_eq!(
456            c.cid_of_order_id("0"),
457            None,
458            "the oldest number is forgotten"
459        );
460        assert_eq!(c.cid_of_order_id(&ORDER_IDS_CAP.to_string()), Some(cid));
461
462        c.register("e", SubscriptionId::new(1002), UdSubKind::Executions);
463        let mut out = EventSink::new();
464        let now = Instant::now();
465        let trade = |n: usize| {
466            json!({"id": n.to_string(), "orderno": "9", "symbol": "SBER", "board": "TQBR", "side": "buy",
467                   "qtyUnits": 1, "price": 10, "date": "2026-10-20T10:00:00Z"})
468        };
469        for n in 0..=SEEN_EXECUTIONS_CAP {
470            c.on_frame(data("e", trade(n)), now, &mut out);
471        }
472        assert_eq!(out.drain().len(), SEEN_EXECUTIONS_CAP + 1);
473        c.on_frame(data("e", trade(0)), now, &mut out);
474        assert_eq!(out.drain().len(), 1, "the oldest trade is told again");
475    }
476
477    #[test]
478    fn the_core_sends_nothing_by_itself_and_has_no_timers() {
479        let mut c = core();
480        assert!(c.poll_transmit().is_none());
481        assert!(c.poll_timeout().is_none());
482        c.on_timeout(Instant::now(), &mut EventSink::new());
483        c.apply(&SubscriptionSet::new());
484    }
485}
brokapi_alor/core/ud.rs

brokapi_alor/core/
ud.rs
