brokapi_bcs/core/
ud.rs

1//! User-data sans-io core.
2
3use std::collections::{HashMap, HashSet, VecDeque};
4use std::time::{Duration, Instant};
5
6use brokapi_bcs_protocol::http::models::{
7    OrderOperationResponse, OrderSearchResponse, OrderStatusResponse,
8};
9use brokapi_bcs_protocol::models::messages::IncomingMessage;
10use brokapi_bcs_protocol::ws::models::OrderMessage;
11use brokapi_core::SubscriptionId;
12use brokapi_sansio::{
13    BrokerCore, CoreEvent, EventSink, OrderRegistry, OrderSignal, Outbound, SubscriptionKind,
14    SubscriptionSet,
15};
16
17use crate::map::execution::{Cumulative, cumulative_of, map_ws_execution};
18use crate::map::order::{map_order_status, map_place_response, map_ws_order_message};
19use crate::map::portfolio::map_ws_portfolio;
20
21/// The answer of a REST call, given to the user data core by hand.
22///
23/// [`UserDataConnector`](crate::UserDataConnector) reads the answers of its own commands directly and
24/// does not send them through the core. These frames exist to drive the core in tests and replays.
25// Boxing the large variant would change the public shape of the enum; these values are short-lived.
26#[allow(clippy::large_enum_variant)]
27#[derive(Debug, Clone)]
28pub enum UdRestFrame {
29    /// The answer to a place request: the order is acknowledged, an `Event::Order` is pushed.
30    Place(OrderOperationResponse),
31    /// The answer to a cancel request: the order is marked cancelled in the core, no event is pushed.
32    Cancel(OrderOperationResponse),
33    /// The answer to a replace request: the new order is acknowledged, no event is pushed.
34    Replace(OrderOperationResponse),
35    /// The answer to an order status read: an `Event::Order` with the reported state is pushed.
36    OrderState(OrderStatusResponse),
37    /// The answer to an active orders search. The core ignores it.
38    ActiveOrders(OrderSearchResponse),
39}
40
41/// A frame for the user data core: a WebSocket push or a REST answer.
42// Boxing the large variant would change the public shape of the enum; these values are short-lived.
43#[allow(clippy::large_enum_variant)]
44#[derive(Debug, Clone)]
45pub enum UdFrame {
46    /// A text frame from one of the WebSocket channels.
47    Ws {
48        /// The channel the frame came from. It decides how the text is parsed.
49        channel: brokapi_bcs_protocol::ws::client::BcsWsChannel,
50        /// The text of the frame, as it came off the socket.
51        text: String,
52    },
53    /// The answer of a REST call.
54    Rest(UdRestFrame),
55}
56
57/// Sans-io user data core: order statuses, fills and portfolio updates from the WebSocket channels.
58///
59/// It turns frames into [`Event`](brokapi_core::Event)s (orders, executions, portfolio), delivers each
60/// fill once even if the broker repeats it (de-duplicated by `executionId`, the last 20 000 are
61/// remembered), and remembers which broker order id belongs to which client order id. It does no I/O.
62///
63/// # Examples
64///
65/// Feed one order event from the `orders/events/ws` channel and read the event it makes:
66///
67/// ```
68/// use std::time::Instant;
69/// use brokapi_bcs_protocol::ws::client::BcsWsChannel;
70/// use brokapi_bcs::UdCore;
71/// use brokapi_core::{ClientOrderId, Event, OrderState, SubscriptionId};
72/// use brokapi_sansio::{
73///     BrokerCore, DesiredSubscription, EventSink, SubscriptionKind, SubscriptionSet,
74/// };
75///
76/// let mut core = UdCore::new();
77/// let sub = SubscriptionId::new(1001);
78/// core.apply(&SubscriptionSet::with_subscriptions(vec![DesiredSubscription {
79///     id: sub,
80///     kind: SubscriptionKind::OrderState,
81/// }]));
82///
83/// let cid = ClientOrderId::new_v4();
84/// let frame = format!(
85///     r#"{{"originalClientOrderId":"{cid}","clientOrderId":"{cid}","data":{{
86///         "messageType":"8","orderStatus":"0","executionType":"0","orderQuantity":10,
87///         "executedQuantity":0,"lastQuantity":0,"remainedQuantity":10,"ticker":"SBER",
88///         "classCode":"TQBR","side":"1","orderType":"2","averagePrice":0,"price":100,
89///         "orderId":"20261019-TQBR-1","currency":"RUB","transactionTime":"2026-10-19T10:00:00.000Z"}}}}"#
90/// );
91/// let mut out = EventSink::new();
92/// core.feed_ws(BcsWsChannel::OrderEvents, &frame, Instant::now(), &mut out);
93///
94/// let events = out.drain();
95/// assert_eq!(events.len(), 1);
96/// assert_eq!(events[0].sub, Some(sub));
97/// let Event::Order(order) = &events[0].payload else { panic!("an order event") };
98/// assert!(matches!(order.state, OrderState::Acked { .. }));
99/// // The core now knows which of our orders the broker's id is.
100/// assert_eq!(core.cid_of_order_id("20261019-TQBR-1"), Some(cid));
101/// ```
102#[derive(Debug)]
103pub struct UdCore {
104    desired: SubscriptionSet,
105    actual: HashMap<SubscriptionId, bool>,
106    outbound: VecDeque<Outbound>,
107    orders: OrderRegistry,
108    command_timeout: Duration,
109    next_deadline: Option<Instant>,
110    /// `executionId`s already delivered (the broker repeats messages, and a reconnect replays them).
111    seen_executions: HashSet<String>,
112    seen_order: VecDeque<String>,
113    /// Broker order id (`orderId`) -> our cid. The list of orders (`POST /orders/search`) carries only
114    /// the broker's id, so this is how a listed order gets its cid back. Learned from place answers,
115    /// status reads and the order-events stream.
116    order_ids: HashMap<String, brokapi_core::ClientOrderId>,
117    /// Running totals of each order from its latest message (the broker order id is the key): the price of
118    /// a fill is the growth of the value over the growth of the quantity.
119    totals: HashMap<String, Cumulative>,
120    totals_order: VecDeque<String>,
121    /// Finished orders (cid, broker order id), to forget what is kept for them after the memory window.
122    done: brokapi_runtime::FinishedWindow<(brokapi_core::ClientOrderId, Option<String>)>,
123    /// How many times the order events session announced a connect. From the second on, the connection
124    /// was lost once and what happened to the orders in the gap is not known.
125    connects: u32,
126}
127
128/// How many `executionId`s are remembered for de-duplication.
129const SEEN_EXECUTIONS_CAP: usize = 20_000;
130/// How many orders keep their running totals.
131const TOTALS_CAP: usize = 20_000;
132
133impl Default for UdCore {
134    fn default() -> Self {
135        Self::new()
136    }
137}
138
139impl UdCore {
140    /// Creates a core with no subscriptions and no known orders. The answer timer for a registered
141    /// place command is 30 seconds (see [`register_place`](Self::register_place)).
142    pub fn new() -> Self {
143        Self {
144            desired: SubscriptionSet::new(),
145            actual: HashMap::new(),
146            outbound: VecDeque::new(),
147            orders: OrderRegistry::new(),
148            command_timeout: Duration::from_secs(30),
149            next_deadline: None,
150            seen_executions: HashSet::new(),
151            seen_order: VecDeque::new(),
152            order_ids: HashMap::new(),
153            totals: HashMap::new(),
154            totals_order: VecDeque::new(),
155            done: brokapi_runtime::FinishedWindow::new(),
156            connects: 0,
157        }
158    }
159
160    /// Remembers that the broker's order `order_id` is our `cid`.
161    pub fn learn_order_id(&mut self, order_id: &str, cid: brokapi_core::ClientOrderId) {
162        if !order_id.is_empty() && !cid.is_nil() {
163            self.order_ids.insert(order_id.to_string(), cid);
164        }
165    }
166
167    /// Our cid of the broker's order `order_id`, if this process has seen it.
168    pub fn cid_of_order_id(&self, order_id: &str) -> Option<brokapi_core::ClientOrderId> {
169        self.order_ids.get(order_id).copied()
170    }
171
172    /// The orders this core tracks (state machine per client order id).
173    pub fn orders(&self) -> &OrderRegistry {
174        &self.orders
175    }
176
177    /// The same registry, for changes. Meant for tests and replays.
178    pub fn orders_mut(&mut self) -> &mut OrderRegistry {
179        &mut self.orders
180    }
181
182    /// Registers a place command that was just sent, and starts the 30 second answer timer for it.
183    ///
184    /// [`UserDataConnector`](crate::UserDataConnector) does not call it (it tracks sent orders in its
185    /// own runtime). It is here to drive the core by hand.
186    pub fn register_place(&mut self, cid: brokapi_core::ClientOrderId, now: Instant) {
187        self.orders.register_place(cid, now);
188        self.next_deadline = Some(now + self.command_timeout);
189    }
190
191    fn handle_ws(
192        &mut self,
193        channel: brokapi_bcs_protocol::ws::client::BcsWsChannel,
194        text: &str,
195        now: Instant,
196        out: &mut EventSink,
197    ) {
198        let msg = brokapi_bcs_protocol::ws::parse_incoming(text, channel);
199        self.handle_incoming(msg, now, out);
200    }
201
202    /// Handles one message already parsed by the protocol layer.
203    ///
204    /// This is what the pump calls. Order messages push `Event::Order`, and a message with a fill
205    /// (`lastQuantity` above zero) also pushes `Event::Execution` once per `executionId`, but only
206    /// while somebody is subscribed to executions. Portfolio messages push `Event::Portfolio`. Other
207    /// messages are skipped.
208    pub fn on_incoming(&mut self, msg: IncomingMessage, now: Instant, out: &mut EventSink) {
209        self.handle_incoming(msg, now, out);
210    }
211
212    fn handle_incoming(&mut self, msg: IncomingMessage, now: Instant, out: &mut EventSink) {
213        match msg {
214            IncomingMessage::OrderExecution(order) => {
215                // The current `orders/events/ws` channel (and the deprecated execution-status):
216                // status changes AND fills. A message is a fill when it has `lastQuantity > 0`.
217                self.push_order(&order, now, out);
218                self.push_execution(&order, now, out);
219            }
220            IncomingMessage::OrderTransaction(order) => self.push_order(&order, now, out),
221            IncomingMessage::Reconnected => self.on_connect(out),
222            IncomingMessage::Portfolio(positions) => {
223                out.push(CoreEvent {
224                    sub: self.find_portfolio_sub(),
225                    payload: map_ws_portfolio(&positions, now),
226                });
227                if let Some(id) = self.find_portfolio_sub() {
228                    self.actual.insert(id, true);
229                }
230            }
231            _ => {}
232        }
233    }
234
235    /// The session of the order events channel is (re)connected. From the second time on the connection
236    /// was lost: tell the consumer on both subscriptions that carry orders and fills (the emitter
237    /// closes the trading gate on it).
238    fn on_connect(&mut self, out: &mut EventSink) {
239        self.connects += 1;
240        if self.connects < 2 {
241            return;
242        }
243        let gap_from = crate::map::unix_nanos(chrono::Utc::now());
244        for sub in [self.find_order_state_sub(), self.find_execution_sub()]
245            .into_iter()
246            .flatten()
247        {
248            out.push(CoreEvent {
249                sub: Some(sub),
250                payload: brokapi_core::Event::Reconnected { sub, gap_from },
251            });
252        }
253    }
254
255    fn push_order(&mut self, order: &OrderMessage, now: Instant, out: &mut EventSink) {
256        if let Some(event) = map_ws_order_message(order, now) {
257            if let brokapi_core::Event::Order(snapshot) = &event {
258                if let Some(id) = &snapshot.broker_id {
259                    self.learn_order_id(id.as_str(), snapshot.cid);
260                }
261                self.apply_order_snapshot(snapshot, now);
262            }
263            out.push(CoreEvent {
264                sub: self.find_order_state_sub(),
265                payload: event,
266            });
267        }
268    }
269
270    /// `Event::Execution` for a message with a fill, once per `executionId`. Nothing is emitted
271    /// while nobody is subscribed to executions. The running totals of the order are kept in any case:
272    /// the price of the next fill is computed from them.
273    fn push_execution(&mut self, order: &OrderMessage, now: Instant, out: &mut EventSink) {
274        let Some(data) = order.data.as_ref() else {
275            return;
276        };
277        let order_id = data.order_id.clone().filter(|s| !s.is_empty());
278        let before = order_id
279            .as_ref()
280            .and_then(|id| self.totals.get(id))
281            .copied();
282        let after = cumulative_of(data);
283        let exec = match self.find_execution_sub() {
284            Some(sub) => map_ws_execution(order, now, before).map(|e| (sub, e)),
285            None => None,
286        };
287        if let (Some(id), Some(after)) = (order_id, after) {
288            self.remember_totals(id, after);
289        }
290        let Some((sub, exec)) = exec else {
291            return;
292        };
293        if !self.seen_executions.insert(exec.trade_id.clone()) {
294            return;
295        }
296        self.seen_order.push_back(exec.trade_id.clone());
297        if self.seen_order.len() > SEEN_EXECUTIONS_CAP
298            && let Some(old) = self.seen_order.pop_front()
299        {
300            self.seen_executions.remove(&old);
301        }
302        out.push(CoreEvent {
303            sub: Some(sub),
304            payload: brokapi_core::Event::Execution(exec),
305        });
306    }
307
308    fn remember_totals(&mut self, order_id: String, totals: Cumulative) {
309        if self.totals.insert(order_id.clone(), totals).is_none() {
310            self.totals_order.push_back(order_id);
311            if self.totals_order.len() > TOTALS_CAP
312                && let Some(old) = self.totals_order.pop_front()
313            {
314                self.totals.remove(&old);
315            }
316        }
317    }
318
319    fn find_execution_sub(&self) -> Option<SubscriptionId> {
320        self.desired
321            .subs()
322            .iter()
323            .find_map(|s| matches!(s.kind, SubscriptionKind::Executions).then_some(s.id))
324    }
325
326    fn handle_rest(&mut self, frame: UdRestFrame, now: Instant, out: &mut EventSink) {
327        match frame {
328            UdRestFrame::Place(resp) => {
329                self.orders
330                    .apply(
331                        resp.client_order_id,
332                        OrderSignal::Ack {
333                            broker_id: brokapi_core::BrokerOrderId::new(
334                                resp.client_order_id.hyphenated().to_string(),
335                            ),
336                        },
337                        now,
338                    )
339                    .ok();
340                out.push(CoreEvent {
341                    sub: None,
342                    payload: map_place_response(&resp, now),
343                });
344            }
345            UdRestFrame::Cancel(resp) => {
346                self.orders
347                    .apply(resp.client_order_id, OrderSignal::Cancelled, now)
348                    .ok();
349            }
350            UdRestFrame::Replace(resp) => {
351                self.orders
352                    .apply(
353                        resp.client_order_id,
354                        OrderSignal::Ack {
355                            broker_id: brokapi_core::BrokerOrderId::new(
356                                resp.client_order_id.hyphenated().to_string(),
357                            ),
358                        },
359                        now,
360                    )
361                    .ok();
362            }
363            UdRestFrame::OrderState(resp) => {
364                if let Some(event) = map_order_status(&resp, now) {
365                    if let brokapi_core::Event::Order(snapshot) = &event {
366                        self.apply_order_snapshot(snapshot, now);
367                    }
368                    out.push(CoreEvent {
369                        sub: self.find_order_state_sub(),
370                        payload: event,
371                    });
372                }
373            }
374            UdRestFrame::ActiveOrders(_resp) => {}
375        }
376    }
377
378    fn apply_order_snapshot(&mut self, snapshot: &brokapi_core::OrderSnapshot, now: Instant) {
379        let signal = match &snapshot.state {
380            brokapi_core::OrderState::Acked { broker_id, .. } => OrderSignal::Ack {
381                broker_id: broker_id.clone(),
382            },
383            brokapi_core::OrderState::PartiallyFilled { filled, left, .. } => {
384                OrderSignal::PartialFill {
385                    filled: *filled,
386                    left: *left,
387                }
388            }
389            brokapi_core::OrderState::Filled { .. } => OrderSignal::Filled,
390            brokapi_core::OrderState::Cancelled { .. } => OrderSignal::Cancelled,
391            brokapi_core::OrderState::Rejected { reason, .. } => OrderSignal::Rejected {
392                reason: reason.clone(),
393            },
394            _ => return,
395        };
396        let _ = self.orders.apply(snapshot.cid, signal, now);
397        if snapshot.state.is_final() {
398            let order_id = snapshot.broker_id.as_ref().map(|b| b.as_str().to_string());
399            self.done.push(now, (snapshot.cid, order_id));
400            for (cid, order_id) in self.done.expired(now) {
401                self.orders.forget_finished(cid);
402                if let Some(id) = order_id {
403                    self.order_ids.remove(&id);
404                    self.totals.remove(&id);
405                }
406            }
407        }
408    }
409
410    fn find_order_state_sub(&self) -> Option<SubscriptionId> {
411        self.desired.subs().iter().find_map(|s| {
412            if matches!(s.kind, SubscriptionKind::OrderState) {
413                Some(s.id)
414            } else {
415                None
416            }
417        })
418    }
419
420    fn find_portfolio_sub(&self) -> Option<SubscriptionId> {
421        self.desired.subs().iter().find_map(|s| {
422            if matches!(s.kind, SubscriptionKind::Portfolio) {
423                Some(s.id)
424            } else {
425                None
426            }
427        })
428    }
429
430    /// Parses one raw text frame of a WebSocket `channel` and handles it like
431    /// [`on_incoming`](Self::on_incoming). A frame that cannot be parsed is skipped, there is no
432    /// error to return.
433    pub fn feed_ws(
434        &mut self,
435        channel: brokapi_bcs_protocol::ws::client::BcsWsChannel,
436        text: &str,
437        now: Instant,
438        out: &mut EventSink,
439    ) {
440        let msg = brokapi_bcs_protocol::ws::parse_incoming(text, channel);
441        self.handle_incoming(msg, now, out);
442    }
443}
444
445impl BrokerCore for UdCore {
446    type Frame = UdFrame;
447
448    fn on_frame(&mut self, frame: Self::Frame, now: Instant, out: &mut EventSink) {
449        match frame {
450            UdFrame::Ws { channel, text } => self.handle_ws(channel, &text, now, out),
451            UdFrame::Rest(rest) => self.handle_rest(rest, now, out),
452        }
453    }
454
455    fn poll_transmit(&mut self) -> Option<Outbound> {
456        self.outbound.pop_front()
457    }
458
459    fn poll_timeout(&self) -> Option<Instant> {
460        self.next_deadline
461    }
462
463    fn on_timeout(&mut self, now: Instant, out: &mut EventSink) {
464        let changed = self.orders.on_global_timeout(now, self.command_timeout);
465        for (cid, state) in changed {
466            out.push(CoreEvent {
467                sub: None,
468                payload: brokapi_core::Event::Order(brokapi_core::OrderSnapshot::new(
469                    cid,
470                    brokapi_core::InstrumentId::new(
471                        brokapi_core::InstrumentKey::new("", ""),
472                        brokapi_core::NativeId::TickerClass {
473                            ticker: String::new(),
474                            class_code: String::new(),
475                        },
476                    ),
477                    state,
478                    brokapi_core::Side::Buy,
479                    brokapi_core::OrderType::Limit,
480                    brokapi_core::Qty::ZERO,
481                    brokapi_core::Timestamps::new(brokapi_core::UnixNanos::ZERO, now),
482                )),
483            });
484        }
485        self.next_deadline = None;
486    }
487
488    fn apply(&mut self, desired: &SubscriptionSet) {
489        self.desired = desired.clone();
490        for sub in desired.subs() {
491            match sub.kind {
492                SubscriptionKind::OrderState
493                | SubscriptionKind::Portfolio
494                | SubscriptionKind::Executions => {
495                    self.actual.insert(sub.id, false);
496                }
497                _ => {}
498            }
499        }
500    }
501}
brokapi_bcs/core/ud.rs

brokapi_bcs/core/
ud.rs
