1pub mod config;
19pub mod ids_generator;
20
21mod settlement;
22
23use std::{
24 cell::RefCell,
25 cmp::min,
26 fmt::Debug,
27 mem,
28 ops::{Add, Sub},
29 rc::Rc,
30};
31
32use indexmap::{IndexMap, IndexSet};
33use jiff::SignedDuration;
34use nautilus_common::{
35 cache::Cache,
36 clock::Clock,
37 messages::execution::{
38 BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
39 },
40 msgbus::{self, MessagingSwitchboard},
41};
42use nautilus_core::{UUID4, UnixNanos, correctness::CorrectnessResult};
43use nautilus_model::{
44 data::{
45 Bar, BarType, InstrumentClose, OrderBookDelta, OrderBookDeltas, OrderBookDepth10,
46 QuoteTick, TradeTick,
47 order::{BookOrder, OrderId},
48 },
49 enums::{
50 AccountType, AggregationSource, AggressorSide, BookAction, BookType, ContingencyType,
51 InstrumentCloseType, LiquiditySide, MarketStatus, MarketStatusAction, OmsType, OrderSide,
52 OrderStatus, OrderType, PriceType, RecordFlag, TimeInForce, TriggerType,
53 },
54 events::{
55 OrderAccepted, OrderCancelRejected, OrderCanceled, OrderEventAny, OrderExpired,
56 OrderFilled, OrderModifyRejected, OrderRejected, OrderSubmitted, OrderTriggered,
57 OrderUpdated,
58 },
59 identifiers::{
60 AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TraderId, Venue,
61 VenueOrderId,
62 },
63 instruments::{Instrument, InstrumentAny},
64 orderbook::{BookLevel, OrderBook},
65 orders::{MarketOrder, Order, OrderAny, OrderCore},
66 position::Position,
67 types::{
68 Currency, Money, Price, Quantity, fixed::FIXED_PRECISION, price::PriceRaw,
69 quantity::QuantityRaw,
70 },
71};
72use ustr::Ustr;
73
74use self::{config::OrderMatchingEngineConfig, ids_generator::IdsGenerator};
75use crate::{
76 matching_core::{MatchAction, OrderMatchingCore, RestingOrder},
77 models::{
78 fee::{FeeModel, FeeModelHandle},
79 fill::{FillModel, FillModelHandle},
80 },
81 protection::protection_price_calculate,
82 trailing::trailing_stop_calculate,
83};
84
85pub struct OrderMatchingEngine {
87 pub venue: Venue,
89 pub instrument: InstrumentAny,
91 pub raw_id: u32,
93 pub book_type: BookType,
95 pub oms_type: OmsType,
97 pub account_type: AccountType,
99 pub market_status: MarketStatus,
101 pub config: OrderMatchingEngineConfig,
103 core: OrderMatchingCore,
104 clock: Rc<RefCell<dyn Clock>>,
105 cache: Rc<RefCell<Cache>>,
106 book: OrderBook,
107 fill_model: FillModelHandle,
108 fee_model: FeeModelHandle,
109 event_handler: Option<Rc<dyn Fn(OrderEventAny)>>,
110 target_bid: Option<Price>,
111 target_ask: Option<Price>,
112 target_last: Option<Price>,
113 last_bar_bid: Option<Bar>,
114 last_bar_ask: Option<Bar>,
115 fill_at_market: bool,
116 execution_bar_types: IndexMap<InstrumentId, BarType>,
117 execution_bar_deltas: IndexMap<BarType, SignedDuration>,
118 account_ids: IndexMap<TraderId, AccountId>,
119 cached_filled_qty: IndexMap<ClientOrderId, Quantity>,
120 post_match_order_ids: IndexSet<ClientOrderId>,
121 ids_generator: IdsGenerator,
122 last_trade_size: Option<Quantity>,
123 trade_consumption: QuantityRaw,
124 bid_consumption: IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
125 ask_consumption: IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
126 queue_pending: IndexMap<ClientOrderId, PriceRaw>,
127 queue_ahead_orders: IndexMap<ClientOrderId, IndexMap<OrderId, QuantityRaw>>,
128 queue_ahead_total: IndexMap<ClientOrderId, (PriceRaw, QuantityRaw)>,
129 queue_ids_by_price: IndexMap<PriceRaw, IndexSet<ClientOrderId>>,
130 queue_excess: IndexMap<ClientOrderId, QuantityRaw>,
131 queue_id_scratch: Vec<ClientOrderId>,
132 queue_pending_scratch: Vec<(ClientOrderId, PriceRaw)>,
133 queue_stale_scratch: Vec<ClientOrderId>,
134 queue_entry_scratch: Vec<(ClientOrderId, QuantityRaw, QuantityRaw)>,
135 prev_bid_price_raw: PriceRaw,
136 prev_bid_size_raw: QuantityRaw,
137 prev_ask_price_raw: PriceRaw,
138 prev_ask_size_raw: QuantityRaw,
139 tob_initialized: bool,
140 last_quote_bid: Option<Price>,
141 last_quote_ask: Option<Price>,
142 precision_mismatch_streak: u32,
143 instrument_close: Option<InstrumentClose>,
144 pending_resolution: bool,
145 expiration_processed: bool,
146 option_settlement_failed: bool,
147 option_settlement_warning: Option<&'static str>,
148 option_expiration_orders_canceled: bool,
149}
150
151impl Debug for OrderMatchingEngine {
152 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
153 f.debug_struct(stringify!(OrderMatchingEngine))
154 .field("venue", &self.venue)
155 .field("instrument", &self.instrument.id())
156 .finish()
157 }
158}
159
160impl OrderMatchingEngine {
161 #[expect(clippy::too_many_arguments)]
163 pub fn new(
164 instrument: InstrumentAny,
165 raw_id: u32,
166 fill_model: FillModelHandle,
167 fee_model: FeeModelHandle,
168 book_type: BookType,
169 oms_type: OmsType,
170 account_type: AccountType,
171 clock: Rc<RefCell<dyn Clock>>,
172 cache: Rc<RefCell<Cache>>,
173 config: OrderMatchingEngineConfig,
174 ) -> Self {
175 let book = OrderBook::new(instrument.id(), book_type);
176 let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
177 core.set_fill_limit_inside_spread(Self::fill_limit_inside_spread_or_false(&fill_model));
178 let ids_generator = IdsGenerator::new(
179 instrument.id().venue,
180 oms_type,
181 raw_id,
182 config.use_random_ids,
183 config.use_position_ids,
184 cache.clone(),
185 );
186
187 Self {
188 venue: instrument.id().venue,
189 instrument,
190 raw_id,
191 fill_model,
192 fee_model,
193 event_handler: None,
194 book_type,
195 oms_type,
196 account_type,
197 clock,
198 cache,
199 book,
200 market_status: MarketStatus::Open,
201 config,
202 core,
203 target_bid: None,
204 target_ask: None,
205 target_last: None,
206 last_bar_bid: None,
207 last_bar_ask: None,
208 fill_at_market: true,
209 execution_bar_types: IndexMap::new(),
210 execution_bar_deltas: IndexMap::new(),
211 account_ids: IndexMap::new(),
212 cached_filled_qty: IndexMap::new(),
213 post_match_order_ids: IndexSet::new(),
214 ids_generator,
215 last_trade_size: None,
216 trade_consumption: 0,
217 bid_consumption: IndexMap::new(),
218 ask_consumption: IndexMap::new(),
219 queue_pending: IndexMap::new(),
220 queue_ahead_orders: IndexMap::new(),
221 queue_ahead_total: IndexMap::new(),
222 queue_ids_by_price: IndexMap::new(),
223 queue_excess: IndexMap::new(),
224 queue_id_scratch: Vec::new(),
225 queue_pending_scratch: Vec::new(),
226 queue_stale_scratch: Vec::new(),
227 queue_entry_scratch: Vec::new(),
228 prev_bid_price_raw: 0,
229 prev_bid_size_raw: 0,
230 prev_ask_price_raw: 0,
231 prev_ask_size_raw: 0,
232 tob_initialized: false,
233 last_quote_bid: None,
234 last_quote_ask: None,
235 precision_mismatch_streak: 0,
236 instrument_close: None,
237 pending_resolution: false,
238 expiration_processed: false,
239 option_settlement_failed: false,
240 option_settlement_warning: None,
241 option_expiration_orders_canceled: false,
242 }
243 }
244
245 pub fn set_event_handler(&mut self, handler: Rc<dyn Fn(OrderEventAny)>) {
252 self.event_handler = Some(handler);
253 }
254
255 fn dispatch_order_event(&self, event: OrderEventAny) {
256 if let Some(handler) = &self.event_handler {
257 handler(event);
258 } else {
259 let endpoint = MessagingSwitchboard::exec_engine_process();
260 msgbus::send_order_event(endpoint, event);
261 }
262 }
263
264 pub fn reset(&mut self) {
270 self.book.reset();
271 self.execution_bar_types.clear();
272 self.execution_bar_deltas.clear();
273 self.account_ids.clear();
274 self.cached_filled_qty.clear();
275 self.post_match_order_ids.clear();
276 self.core.reset();
277 self.target_bid = None;
278 self.target_ask = None;
279 self.target_last = None;
280 self.last_trade_size = None;
281 self.trade_consumption = 0;
282 self.bid_consumption.clear();
283 self.ask_consumption.clear();
284 self.queue_pending.clear();
285 self.queue_ahead_orders.clear();
286 self.queue_ahead_total.clear();
287 self.queue_ids_by_price.clear();
288 self.queue_excess.clear();
289 self.queue_id_scratch.clear();
290 self.queue_pending_scratch.clear();
291 self.queue_stale_scratch.clear();
292 self.queue_entry_scratch.clear();
293 self.prev_bid_price_raw = 0;
294 self.prev_bid_size_raw = 0;
295 self.prev_ask_price_raw = 0;
296 self.prev_ask_size_raw = 0;
297 self.tob_initialized = false;
298 self.last_quote_bid = None;
299 self.last_quote_ask = None;
300 self.last_bar_bid = None;
301 self.last_bar_ask = None;
302 self.precision_mismatch_streak = 0;
303 self.instrument_close = None;
304 self.market_status = MarketStatus::Open;
305 self.pending_resolution = false;
306 self.expiration_processed = false;
307 self.option_settlement_failed = false;
308 self.option_settlement_warning = None;
309 self.option_expiration_orders_canceled = false;
310 self.fill_at_market = true;
311 self.ids_generator.reset();
312
313 log::info!("Reset {}", self.instrument.id());
314 }
315
316 fn apply_liquidity_consumption(
317 &mut self,
318 mut fills: Vec<(Price, Quantity)>,
319 order_side: OrderSide,
320 leaves_qty: Quantity,
321 book_prices: Option<&[Price]>,
322 ) -> Vec<(Price, Quantity)> {
323 if !self.config.liquidity_consumption {
324 return fills;
325 }
326
327 let consumption = match order_side {
328 OrderSide::Buy => &mut self.ask_consumption,
329 OrderSide::Sell => &mut self.bid_consumption,
330 };
331
332 let mut adjusted_len = 0;
333 let mut remaining_qty = leaves_qty.raw;
334
335 for fill_idx in 0..fills.len() {
336 if remaining_qty == 0 {
337 break;
338 }
339
340 let (price, qty) = fills[fill_idx];
341
342 let book_price = book_prices
345 .and_then(|bp| bp.get(fill_idx).copied())
346 .unwrap_or(price);
347
348 let book_price_raw = book_price.raw;
349 let level_size = self
350 .book
351 .get_quantity_at_level(book_price, order_side, qty.precision);
352
353 let (original_size, consumed) = consumption
354 .entry(book_price_raw)
355 .or_insert((level_size.raw, 0));
356
357 if *original_size != level_size.raw {
359 *original_size = level_size.raw;
360 *consumed = 0;
361 }
362
363 let available = original_size.saturating_sub(*consumed);
364 if available == 0 {
365 continue;
366 }
367
368 let adjusted_qty_raw = min(min(qty.raw, available), remaining_qty);
369 if adjusted_qty_raw == 0 {
370 continue;
371 }
372
373 *consumed += adjusted_qty_raw;
374 remaining_qty -= adjusted_qty_raw;
375
376 let adjusted_qty = Quantity::from_raw(adjusted_qty_raw, qty.precision);
377 fills[adjusted_len] = (price, adjusted_qty);
378 adjusted_len += 1;
379 }
380
381 fills.truncate(adjusted_len);
382 fills
383 }
384
385 fn seed_trade_consumption(
386 &mut self,
387 trade_price_raw: PriceRaw,
388 trade_size_raw: QuantityRaw,
389 trade_ts_event: UnixNanos,
390 aggressor_side: AggressorSide,
391 ) {
392 if trade_size_raw == 0 {
393 return;
394 }
395
396 if self.book.ts_last > trade_ts_event {
399 return;
400 }
401
402 let book = &self.book;
403 let consumption = match aggressor_side {
404 AggressorSide::Buy => &mut self.ask_consumption,
405 AggressorSide::Sell => &mut self.bid_consumption,
406 AggressorSide::NoAggressor => return,
407 };
408
409 let mut remaining = trade_size_raw;
410
411 match aggressor_side {
412 AggressorSide::Buy => {
413 for level in book
414 .asks(None)
415 .take_while(|level| level.price.value.raw <= trade_price_raw)
416 {
417 Self::consume_trade_level(consumption, &mut remaining, level);
418 if remaining == 0 {
419 break;
420 }
421 }
422 }
423 AggressorSide::Sell => {
424 for level in book
425 .bids(None)
426 .take_while(|level| level.price.value.raw >= trade_price_raw)
427 {
428 Self::consume_trade_level(consumption, &mut remaining, level);
429 if remaining == 0 {
430 break;
431 }
432 }
433 }
434 AggressorSide::NoAggressor => unreachable!(),
435 }
436 }
437
438 fn consume_trade_level(
439 consumption: &mut IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
440 remaining: &mut QuantityRaw,
441 level: &BookLevel,
442 ) {
443 let level_size = level.size_raw();
444 let entry = consumption
445 .entry(level.price.value.raw)
446 .or_insert((level_size, 0));
447
448 if entry.0 != level_size {
450 entry.0 = level_size;
451 entry.1 = 0;
452 }
453
454 let available = level_size.saturating_sub(entry.1);
455 let consume = min(*remaining, available);
456 entry.1 += consume;
457 *remaining -= consume;
458 }
459
460 pub fn set_fill_model(&mut self, fill_model: FillModelHandle) {
462 self.core
463 .set_fill_limit_inside_spread(Self::fill_limit_inside_spread_or_false(&fill_model));
464 self.fill_model = fill_model;
465 }
466
467 fn fill_limit_inside_spread_or_false(fill_model: &FillModelHandle) -> bool {
468 fill_model.fill_limit_inside_spread().unwrap_or_else(|e| {
469 log::error!("Failed to query fill model spread behavior: {e}");
470 false
471 })
472 }
473
474 fn snapshot_queue_position(&mut self, order: &OrderAny, price: Price) {
475 if !self.config.queue_position {
476 return;
477 }
478 let size_prec = self.instrument.size_precision();
479
480 let qty_ahead = self.book.get_quantity_at_level(
483 price,
484 OrderCore::opposite_side(order.order_side()),
485 size_prec,
486 );
487
488 let client_order_id = order.client_order_id();
489
490 self.remove_queue_position(client_order_id);
491 self.queue_ids_by_price
492 .entry(price.raw)
493 .or_default()
494 .insert(client_order_id);
495
496 if self.book_type == BookType::L1_MBP && qty_ahead.raw == 0 {
501 let behind_bbo = match order.order_side() {
502 OrderSide::Buy => self.book.best_bid_price().is_some_and(|bid| price < bid),
503 OrderSide::Sell => self.book.best_ask_price().is_some_and(|ask| price > ask),
504 };
505
506 if behind_bbo {
507 self.queue_pending.insert(client_order_id, price.raw);
508 return;
509 }
510 }
511
512 self.queue_ahead_total
513 .insert(client_order_id, (price.raw, qty_ahead.raw));
514
515 if self.book_type == BookType::L3_MBO {
517 let orders_ahead: IndexMap<OrderId, QuantityRaw> = self
518 .book
519 .get_orders_at_level(price, OrderCore::opposite_side(order.order_side()))
520 .iter()
521 .map(|book_order| (book_order.order_id, book_order.size.raw))
522 .collect();
523 self.queue_ahead_orders
524 .insert(client_order_id, orders_ahead);
525 }
526 }
527
528 fn remove_queue_position(&mut self, client_order_id: ClientOrderId) {
529 let pending_price = self.queue_pending.shift_remove(&client_order_id);
530 let ahead_price = self
531 .queue_ahead_total
532 .shift_remove(&client_order_id)
533 .map(|(price_raw, _)| price_raw);
534 self.queue_ahead_orders.shift_remove(&client_order_id);
535 self.queue_excess.shift_remove(&client_order_id);
536
537 for price_raw in [pending_price, ahead_price].into_iter().flatten() {
538 let remove_price = self
539 .queue_ids_by_price
540 .get_mut(&price_raw)
541 .is_some_and(|ids| {
542 ids.shift_remove(&client_order_id);
543 ids.is_empty()
544 });
545
546 if remove_price {
547 self.queue_ids_by_price.shift_remove(&price_raw);
548 }
549 }
550 }
551
552 fn take_queue_ids_at_price(&mut self, price_raw: PriceRaw) -> Vec<ClientOrderId> {
553 let mut ids = Self::take_cleared(&mut self.queue_id_scratch);
554 if let Some(tracked_ids) = self.queue_ids_by_price.get(&price_raw) {
555 ids.extend(tracked_ids.iter().copied());
556 }
557
558 ids
559 }
560
561 fn decrement_queue_on_trade(
562 &mut self,
563 price_raw: PriceRaw,
564 trade_size_raw: QuantityRaw,
565 aggressor_side: AggressorSide,
566 ) {
567 if !self.config.queue_position {
568 return;
569 }
570
571 self.queue_excess.clear();
572
573 let keys = self.take_queue_ids_at_price(price_raw);
574 let mut entries = Self::take_cleared(&mut self.queue_entry_scratch);
575 let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
576
577 for client_order_id in keys.iter().copied() {
578 let (order_price_raw, ahead_raw) =
579 match self.queue_ahead_total.get(&client_order_id).copied() {
580 Some(v) => v,
581 None => continue,
582 };
583
584 let cache = self.cache.borrow();
585 let order_info = cache.order(&client_order_id).and_then(|order| {
586 if order.is_closed() {
587 None
588 } else {
589 Some((order.order_side(), order.leaves_qty().raw))
590 }
591 });
592 drop(cache);
593
594 let Some((order_side, leaves_raw)) = order_info else {
595 stale.push(client_order_id);
596 continue;
597 };
598
599 if order_price_raw != price_raw || ahead_raw == 0 {
600 continue;
601 }
602
603 let should_decrement = matches!(aggressor_side, AggressorSide::NoAggressor)
604 || (aggressor_side == AggressorSide::Buy && order_side == OrderSide::Sell)
605 || (aggressor_side == AggressorSide::Sell && order_side == OrderSide::Buy);
606
607 if should_decrement {
608 entries.push((client_order_id, ahead_raw, leaves_raw));
609 }
610 }
611
612 for id in stale.drain(..) {
613 self.remove_queue_position(id);
614 }
615
616 entries.sort_by_key(|&(_, ahead, _)| ahead);
618
619 let mut remaining = trade_size_raw;
620 let mut prev_position: QuantityRaw = 0;
621
622 for (client_order_id, ahead_raw, leaves_raw) in &entries {
623 if remaining == 0 {
624 let new_ahead = ahead_raw.saturating_sub(trade_size_raw);
625 self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, new_ahead);
626 if new_ahead == 0 {
627 self.queue_excess.insert(*client_order_id, 0);
629 }
630 continue;
631 }
632
633 let gap = ahead_raw.saturating_sub(prev_position);
635 let queue_consumed = remaining.min(gap);
636 remaining -= queue_consumed;
637
638 if remaining == 0 && queue_consumed < gap {
639 let new_ahead = ahead_raw.saturating_sub(trade_size_raw);
640 self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, new_ahead);
641 continue;
642 }
643
644 self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, 0);
645 let excess = remaining.min(*leaves_raw);
646 self.queue_excess.insert(*client_order_id, excess);
647 remaining -= excess;
648 prev_position = ahead_raw + excess;
649 }
650
651 self.queue_id_scratch = keys;
652 self.queue_entry_scratch = entries;
653 self.queue_stale_scratch = stale;
654 }
655
656 fn reduce_queue_ahead(
660 &mut self,
661 client_order_id: ClientOrderId,
662 price_raw: PriceRaw,
663 ahead_raw: QuantityRaw,
664 new_ahead_raw: QuantityRaw,
665 ) {
666 self.queue_ahead_total
667 .insert(client_order_id, (price_raw, new_ahead_raw));
668 self.consume_queue_ahead_orders(client_order_id, ahead_raw.saturating_sub(new_ahead_raw));
669 }
670
671 fn consume_queue_ahead_orders(
673 &mut self,
674 client_order_id: ClientOrderId,
675 mut amount_raw: QuantityRaw,
676 ) {
677 let Some(orders_ahead) = self.queue_ahead_orders.get_mut(&client_order_id) else {
678 return;
679 };
680
681 while amount_raw > 0 {
682 let Some((&book_order_id, &size_raw)) = orders_ahead.get_index(0) else {
683 break;
684 };
685
686 if size_raw <= amount_raw {
687 orders_ahead.shift_remove(&book_order_id);
688 amount_raw -= size_raw;
689 } else {
690 orders_ahead.insert(book_order_id, size_raw - amount_raw);
691 amount_raw = 0;
692 }
693 }
694 }
695
696 fn determine_trade_fill_qty(&self, order: &OrderAny) -> Option<QuantityRaw> {
697 if !self.config.queue_position {
698 return Some(order.leaves_qty().raw);
699 }
700
701 let client_order_id = order.client_order_id();
702
703 if self.queue_pending.contains_key(&client_order_id) {
705 return None;
706 }
707
708 if let Some(&(tracked_price_raw, ahead_raw)) = self.queue_ahead_total.get(&client_order_id)
709 && let Some(order_price) = order.price()
710 && order_price.raw == tracked_price_raw
711 && ahead_raw > 0
712 {
713 return None;
714 }
715
716 let leaves_raw = order.leaves_qty().raw;
717 if leaves_raw == 0 {
718 return None;
719 }
720
721 let mut available_raw = leaves_raw;
722
723 if let Some(trade_size) = self.last_trade_size {
725 let remaining = trade_size.raw.saturating_sub(self.trade_consumption);
726 available_raw = available_raw.min(remaining);
727
728 if let Some(&excess_raw) = self.queue_excess.get(&client_order_id) {
729 if excess_raw == 0 {
730 return None;
731 }
732 available_raw = available_raw.min(excess_raw);
733 }
734 }
735
736 if available_raw == 0 {
737 return None;
738 }
739
740 Some(available_raw)
741 }
742
743 fn clear_all_queue_positions(&mut self) {
744 for (_, (_, ahead_raw)) in &mut self.queue_ahead_total {
745 *ahead_raw = 0;
746 }
747
748 for orders_ahead in self.queue_ahead_orders.values_mut() {
749 orders_ahead.clear();
750 }
751 }
752
753 fn adjust_queue_for_delta(&mut self, delta: &OrderBookDelta) {
754 if delta.action == BookAction::Delete {
755 if self.is_order_granular_delta(delta.flags) {
756 self.advance_l3_queue_on_delete(delta.order.order_id);
757 } else {
758 self.clear_queue_on_delete(delta.order.price.raw, delta.order.side);
759 }
760 } else if delta.action == BookAction::Update {
761 if self.is_order_granular_delta(delta.flags) {
762 self.adjust_l3_queue_on_update(&delta.order);
763 } else {
764 self.cap_queue_ahead(
765 delta.order.price.raw,
766 delta.order.size.raw,
767 delta.order.side,
768 );
769 }
770 }
771 }
772
773 fn clear_queue_on_delete(
774 &mut self,
775 deleted_price_raw: PriceRaw,
776 deleted_side: Option<OrderSide>,
777 ) {
778 let keys = self.take_queue_ids_at_price(deleted_price_raw);
779 for client_order_id in keys.iter().copied() {
780 if let Some(&(order_price_raw, ahead_raw)) =
781 self.queue_ahead_total.get(&client_order_id)
782 && order_price_raw == deleted_price_raw
783 {
784 let matches_side = self
785 .cache
786 .borrow()
787 .order(&client_order_id)
788 .is_some_and(|o| Some(o.order_side()) == deleted_side);
789
790 if matches_side {
791 self.reduce_queue_ahead(client_order_id, order_price_raw, ahead_raw, 0);
792 }
793 }
794 }
795
796 self.queue_id_scratch = keys;
797 }
798
799 fn is_order_granular_delta(&self, flags: u8) -> bool {
802 self.book_type == BookType::L3_MBO
803 && !RecordFlag::F_TOB.matches(flags)
804 && !RecordFlag::F_MBP.matches(flags)
805 }
806
807 fn advance_l3_queue_on_delete(&mut self, book_order_id: OrderId) {
808 for (client_order_id, orders_ahead) in &mut self.queue_ahead_orders {
809 let Some(size_raw) = orders_ahead.shift_remove(&book_order_id) else {
810 continue;
811 };
812
813 if let Some((_, ahead_raw)) = self.queue_ahead_total.get_mut(client_order_id) {
814 *ahead_raw = ahead_raw.saturating_sub(size_raw);
815 }
816 }
817 }
818
819 fn adjust_l3_queue_on_update(&mut self, book_order: &BookOrder) {
825 for (client_order_id, orders_ahead) in &mut self.queue_ahead_orders {
826 let Some(&tracked_size_raw) = orders_ahead.get(&book_order.order_id) else {
827 continue;
828 };
829 let Some((tracked_price_raw, ahead_raw)) =
830 self.queue_ahead_total.get_mut(client_order_id)
831 else {
832 continue;
833 };
834
835 if book_order.price.raw != *tracked_price_raw {
836 *ahead_raw = ahead_raw.saturating_sub(tracked_size_raw);
837 orders_ahead.shift_remove(&book_order.order_id);
838 } else if book_order.size.raw < tracked_size_raw {
839 *ahead_raw = ahead_raw.saturating_sub(tracked_size_raw - book_order.size.raw);
841 orders_ahead.insert(book_order.order_id, book_order.size.raw);
842 } else if book_order.size.raw > tracked_size_raw {
843 *ahead_raw = ahead_raw.saturating_add(book_order.size.raw - tracked_size_raw);
844 orders_ahead.insert(book_order.order_id, book_order.size.raw);
845 }
846 }
847 }
848
849 fn cap_queue_ahead(
850 &mut self,
851 price_raw: PriceRaw,
852 size_raw: QuantityRaw,
853 order_side: Option<OrderSide>,
854 ) {
855 let keys = self.take_queue_ids_at_price(price_raw);
856 let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
857
858 for client_order_id in keys.iter().copied() {
859 let (order_price_raw, ahead_raw) =
860 match self.queue_ahead_total.get(&client_order_id).copied() {
861 Some(v) => v,
862 None => continue,
863 };
864
865 if order_price_raw != price_raw || ahead_raw <= size_raw {
866 continue;
867 }
868
869 let cache = self.cache.borrow();
870 let order_info = cache.order(&client_order_id).and_then(|order| {
871 if order.is_closed() {
872 None
873 } else {
874 Some(order.order_side())
875 }
876 });
877 drop(cache);
878
879 let Some(side) = order_info else {
880 stale.push(client_order_id);
881 continue;
882 };
883
884 if Some(side) != order_side {
885 continue;
886 }
887
888 self.reduce_queue_ahead(client_order_id, order_price_raw, ahead_raw, size_raw);
889 }
890
891 for id in stale.drain(..) {
892 self.remove_queue_position(id);
893 }
894
895 self.queue_id_scratch = keys;
896 self.queue_stale_scratch = stale;
897 }
898
899 fn seed_tob_baseline(&mut self) {
900 let bid = self.book.best_bid_price();
901 let ask = self.book.best_ask_price();
902 self.prev_bid_price_raw = bid.map_or(0, |p| p.raw);
903 self.prev_bid_size_raw = self.book.best_bid_size().map_or(0, |q| q.raw);
904 self.prev_ask_price_raw = ask.map_or(0, |p| p.raw);
905 self.prev_ask_size_raw = self.book.best_ask_size().map_or(0, |q| q.raw);
906 self.tob_initialized = bid.is_some() || ask.is_some();
907 }
908
909 fn decrement_l1_queue_on_quote(
910 &mut self,
911 bid_price_raw: PriceRaw,
912 bid_size_raw: QuantityRaw,
913 ask_price_raw: PriceRaw,
914 ask_size_raw: QuantityRaw,
915 ) {
916 if !self.config.queue_position {
917 return;
918 }
919
920 if self.tob_initialized {
922 if bid_price_raw < self.prev_bid_price_raw {
924 self.adjust_l1_queue_on_price_move(bid_price_raw, bid_size_raw, OrderSide::Buy);
925 }
926
927 if ask_price_raw > self.prev_ask_price_raw {
929 self.adjust_l1_queue_on_price_move(ask_price_raw, ask_size_raw, OrderSide::Sell);
930 }
931 }
932
933 self.resolve_pending_l1_snapshots(bid_price_raw, bid_size_raw, ask_price_raw, ask_size_raw);
935 }
936
937 fn adjust_l1_queue_on_price_move(
938 &mut self,
939 new_price_raw: PriceRaw,
940 new_size_raw: QuantityRaw,
941 order_side: OrderSide,
942 ) {
943 let mut keys = Self::take_cleared(&mut self.queue_id_scratch);
944 keys.extend(self.queue_ahead_total.keys().copied());
945 let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
946
947 for client_order_id in keys.iter().copied() {
948 let Some(&(order_price_raw, ahead_raw)) = self.queue_ahead_total.get(&client_order_id)
949 else {
950 continue;
951 };
952
953 let cache = self.cache.borrow();
954 let order_info = cache.order(&client_order_id).and_then(|order| {
955 if order.is_closed() {
956 None
957 } else {
958 Some(order.order_side())
959 }
960 });
961 drop(cache);
962
963 let Some(side) = order_info else {
964 stale.push(client_order_id);
965 continue;
966 };
967
968 if side != order_side {
969 continue;
970 }
971
972 let crossed = match order_side {
975 OrderSide::Buy => order_price_raw > new_price_raw,
976 _ => order_price_raw < new_price_raw,
977 };
978
979 if crossed {
980 self.queue_ahead_total
981 .insert(client_order_id, (order_price_raw, 0));
982 } else if order_price_raw == new_price_raw && ahead_raw > new_size_raw {
983 self.queue_ahead_total
984 .insert(client_order_id, (order_price_raw, new_size_raw));
985 }
986 }
987
988 for id in stale.drain(..) {
989 self.remove_queue_position(id);
990 }
991
992 let mut pending = Self::take_cleared(&mut self.queue_pending_scratch);
993 pending.extend(
994 self.queue_pending
995 .iter()
996 .map(|(&client_order_id, &price_raw)| (client_order_id, price_raw)),
997 );
998
999 for (client_order_id, order_price_raw) in pending.iter().copied() {
1000 let cache = self.cache.borrow();
1001 let order_info = cache.order(&client_order_id).and_then(|order| {
1002 if order.is_closed() {
1003 None
1004 } else {
1005 Some(order.order_side())
1006 }
1007 });
1008 drop(cache);
1009
1010 let Some(side) = order_info else {
1011 stale.push(client_order_id);
1012 continue;
1013 };
1014
1015 if side != order_side {
1016 continue;
1017 }
1018
1019 let crossed = match order_side {
1020 OrderSide::Buy => order_price_raw > new_price_raw,
1021 _ => order_price_raw < new_price_raw,
1022 };
1023
1024 if crossed {
1025 self.queue_pending.shift_remove(&client_order_id);
1026 self.queue_ahead_total
1027 .insert(client_order_id, (order_price_raw, 0));
1028 } else if order_price_raw == new_price_raw {
1029 self.queue_pending.shift_remove(&client_order_id);
1030 self.queue_ahead_total
1031 .insert(client_order_id, (order_price_raw, new_size_raw));
1032 }
1033 }
1034
1035 for id in stale.drain(..) {
1036 self.remove_queue_position(id);
1037 }
1038
1039 self.queue_id_scratch = keys;
1040 self.queue_pending_scratch = pending;
1041 self.queue_stale_scratch = stale;
1042 }
1043
1044 fn resolve_pending_l1_snapshots(
1045 &mut self,
1046 bid_price_raw: PriceRaw,
1047 bid_size_raw: QuantityRaw,
1048 ask_price_raw: PriceRaw,
1049 ask_size_raw: QuantityRaw,
1050 ) {
1051 let mut keys = self.take_queue_ids_at_price(bid_price_raw);
1052 if ask_price_raw != bid_price_raw
1053 && let Some(ask_ids) = self.queue_ids_by_price.get(&ask_price_raw)
1054 {
1055 keys.extend(ask_ids.iter().copied());
1056 }
1057
1058 let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
1059
1060 for client_order_id in keys.iter().copied() {
1061 let Some(&order_price_raw) = self.queue_pending.get(&client_order_id) else {
1062 continue;
1063 };
1064
1065 let cache = self.cache.borrow();
1066 let order_info = cache.order(&client_order_id).and_then(|order| {
1067 if order.is_closed() {
1068 None
1069 } else {
1070 Some(order.order_side())
1071 }
1072 });
1073 drop(cache);
1074
1075 let Some(side) = order_info else {
1076 stale.push(client_order_id);
1077 continue;
1078 };
1079
1080 let matched_size = match side {
1082 OrderSide::Buy if order_price_raw == bid_price_raw => Some(bid_size_raw),
1083 OrderSide::Sell if order_price_raw == ask_price_raw => Some(ask_size_raw),
1084 _ => None,
1085 };
1086
1087 if let Some(size) = matched_size {
1088 self.queue_pending.shift_remove(&client_order_id);
1089 self.queue_ahead_total
1090 .insert(client_order_id, (order_price_raw, size));
1091 }
1092 }
1093
1094 for id in stale.drain(..) {
1095 self.remove_queue_position(id);
1096 }
1097
1098 self.queue_id_scratch = keys;
1099 self.queue_stale_scratch = stale;
1100 }
1101
1102 fn resolve_pending_on_trade(&mut self, trade_price_raw: PriceRaw) {
1103 let mut keys = Self::take_cleared(&mut self.queue_id_scratch);
1104 keys.extend(self.queue_pending.keys().copied());
1105 let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
1106
1107 for client_order_id in keys.iter().copied() {
1108 let Some(&order_price_raw) = self.queue_pending.get(&client_order_id) else {
1109 continue;
1110 };
1111
1112 let cache = self.cache.borrow();
1113 let order_side = cache.order(&client_order_id).and_then(|order| {
1114 if order.is_closed() {
1115 None
1116 } else {
1117 Some(order.order_side())
1118 }
1119 });
1120 drop(cache);
1121
1122 let Some(side) = order_side else {
1123 stale.push(client_order_id);
1124 continue;
1125 };
1126
1127 let crossed = match side {
1129 OrderSide::Buy => trade_price_raw < order_price_raw,
1130 OrderSide::Sell => trade_price_raw > order_price_raw,
1131 };
1132
1133 if crossed {
1134 self.queue_pending.shift_remove(&client_order_id);
1135 self.queue_ahead_total
1136 .insert(client_order_id, (order_price_raw, 0));
1137 }
1138 }
1139
1140 for id in stale.drain(..) {
1141 self.remove_queue_position(id);
1142 }
1143
1144 self.queue_id_scratch = keys;
1145 self.queue_stale_scratch = stale;
1146 }
1147
1148 fn take_cleared<T>(buf: &mut Vec<T>) -> Vec<T> {
1149 let mut items = mem::take(buf);
1150 items.clear();
1151 items
1152 }
1153
1154 #[must_use]
1155 pub fn best_bid_price(&self) -> Option<Price> {
1157 self.book.best_bid_price()
1158 }
1159
1160 #[must_use]
1161 pub fn best_ask_price(&self) -> Option<Price> {
1163 self.book.best_ask_price()
1164 }
1165
1166 #[must_use]
1167 pub const fn get_book(&self) -> &OrderBook {
1169 &self.book
1170 }
1171
1172 #[must_use]
1173 pub fn get_open_bid_orders(&self) -> Vec<RestingOrder> {
1175 self.core.get_orders_bid()
1176 }
1177
1178 #[must_use]
1179 pub fn get_open_ask_orders(&self) -> Vec<RestingOrder> {
1181 self.core.get_orders_ask()
1182 }
1183
1184 #[must_use]
1185 pub fn get_open_orders(&self) -> Vec<RestingOrder> {
1187 self.core.get_orders()
1188 }
1189
1190 #[must_use]
1191 pub fn order_exists(&self, client_order_id: ClientOrderId) -> bool {
1193 self.core.order_exists(client_order_id)
1194 }
1195
1196 #[must_use]
1197 pub fn cached_filled_qty_len(&self) -> usize {
1199 self.cached_filled_qty.len()
1200 }
1201
1202 #[must_use]
1203 pub const fn get_core(&self) -> &OrderMatchingCore {
1204 &self.core
1205 }
1206
1207 pub fn set_fill_at_market(&mut self, value: bool) {
1208 self.fill_at_market = value;
1209 }
1210
1211 pub fn update_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
1217 if instrument.id() != self.instrument.id() {
1218 anyhow::bail!(
1219 "Cannot update instrument {} with {}",
1220 self.instrument.id(),
1221 instrument.id()
1222 );
1223 }
1224
1225 let changed = instrument.price_increment() != self.instrument.price_increment()
1226 || instrument.price_precision() != self.instrument.price_precision()
1227 || instrument.size_precision() != self.instrument.size_precision();
1228
1229 if changed {
1230 self.core
1231 .update_price_increment(instrument.price_increment());
1232 self.book.reset();
1233 self.trade_consumption = 0;
1234 self.bid_consumption.clear();
1235 self.ask_consumption.clear();
1236 self.queue_pending.clear();
1237 self.queue_ahead_orders.clear();
1238 self.queue_ahead_total.clear();
1239 self.queue_ids_by_price.clear();
1240 self.queue_excess.clear();
1241 self.prev_bid_price_raw = 0;
1242 self.prev_bid_size_raw = 0;
1243 self.prev_ask_price_raw = 0;
1244 self.prev_ask_size_raw = 0;
1245 self.tob_initialized = false;
1246 self.last_quote_bid = None;
1247 self.last_quote_ask = None;
1248 self.precision_mismatch_streak = 0;
1249 self.target_bid = None;
1250 self.target_ask = None;
1251 self.target_last = None;
1252 self.last_bar_bid = None;
1253 self.last_bar_ask = None;
1254 self.core.bid = None;
1255 self.core.ask = None;
1256 self.core.last = None;
1257 log::info!(
1258 "Updated instrument {} (price_precision={} size_precision={})",
1259 instrument.id(),
1260 instrument.price_precision(),
1261 instrument.size_precision()
1262 );
1263 }
1264
1265 self.instrument = instrument;
1266
1267 if changed {
1268 self.drop_incompatible_core_orders();
1269 }
1270
1271 Ok(())
1272 }
1273
1274 fn check_price_precision(&self, actual: u8, field: &str) -> anyhow::Result<()> {
1275 let expected = self.instrument.price_precision();
1276 if actual != expected {
1277 anyhow::bail!(
1278 "Invalid {field} precision {actual}, expected {expected} for {}",
1279 self.instrument.id()
1280 );
1281 }
1282 Ok(())
1283 }
1284
1285 fn check_size_precision(&self, actual: u8, field: &str) -> anyhow::Result<()> {
1286 let expected = self.instrument.size_precision();
1287 if actual != expected {
1288 anyhow::bail!(
1289 "Invalid {field} precision {actual}, expected {expected} for {}",
1290 self.instrument.id()
1291 );
1292 }
1293 Ok(())
1294 }
1295
1296 fn log_precision_mismatch(
1297 &mut self,
1298 data_type: &str,
1299 instrument_id: InstrumentId,
1300 err: &anyhow::Error,
1301 ) {
1302 self.precision_mismatch_streak = self.precision_mismatch_streak.saturating_add(1);
1303 let streak = self.precision_mismatch_streak;
1304
1305 if streak <= 3 || streak.is_multiple_of(100) {
1306 log::warn!(
1307 "Skipping {data_type} for {instrument_id}: {err} \
1308 (consecutive_precision_mismatches={streak})"
1309 );
1310 }
1311
1312 if streak == 20 {
1313 log::error!(
1314 "Precision mismatches reached {streak} consecutive events for \
1315 {instrument_id}; check instrument update flow and upstream market data"
1316 );
1317 }
1318 }
1319
1320 fn drop_incompatible_core_orders(&mut self) {
1321 let client_order_ids: Vec<ClientOrderId> = self
1322 .core
1323 .iter_orders()
1324 .filter(|order| {
1325 !self.resting_order_matches_current_instrument(order)
1326 || !self.cached_order_matches_current_instrument(order.client_order_id)
1327 })
1328 .map(|order| order.client_order_id)
1329 .collect();
1330
1331 for client_order_id in client_order_ids {
1332 let order = self
1333 .cache
1334 .borrow()
1335 .order(&client_order_id)
1336 .map(|o| o.clone());
1337
1338 if let Some(order) = order
1339 && (order.is_inflight() || order.is_open())
1340 {
1341 log::warn!(
1342 "Canceling order {client_order_id} after instrument update: \
1343 price, trigger price, or quantity is not compatible with {}",
1344 self.instrument.id()
1345 );
1346 self.cancel_order(&order, None);
1347 } else {
1348 self.delete_core_order(client_order_id);
1349 self.cached_filled_qty.swap_remove(&client_order_id);
1350 }
1351 }
1352 }
1353
1354 fn cached_order_matches_current_instrument(&self, client_order_id: ClientOrderId) -> bool {
1355 self.cache
1356 .borrow()
1357 .order(&client_order_id)
1358 .is_none_or(|order| {
1359 Self::quantity_matches_precision(order.quantity(), self.instrument.size_precision())
1360 })
1361 }
1362
1363 fn resting_order_matches_current_instrument(&self, order: &RestingOrder) -> bool {
1364 order
1365 .limit_price
1366 .is_none_or(|price| self.price_matches_current_instrument(price))
1367 && order
1368 .trigger_price
1369 .is_none_or(|price| self.price_matches_current_instrument(price))
1370 }
1371
1372 fn price_matches_current_instrument(&self, price: Price) -> bool {
1373 Self::price_matches_precision(price, self.instrument.price_precision())
1374 && Self::price_matches_tick(price, self.instrument.price_increment())
1375 }
1376
1377 fn price_matches_precision(price: Price, precision: u8) -> bool {
1378 let precision_diff = FIXED_PRECISION.saturating_sub(precision);
1379 let scale = PriceRaw::pow(10, u32::from(precision_diff));
1380 price.raw % scale == 0
1381 }
1382
1383 fn price_matches_tick(price: Price, increment: Price) -> bool {
1384 let increment_raw = increment.raw.abs();
1385 increment_raw == 0 || price.raw % increment_raw == 0
1386 }
1387
1388 fn quantity_matches_precision(quantity: Quantity, precision: u8) -> bool {
1389 let precision_diff = FIXED_PRECISION.saturating_sub(precision);
1390 let scale = QuantityRaw::pow(10, u32::from(precision_diff));
1391 quantity.raw.is_multiple_of(scale)
1392 }
1393
1394 fn normalize_price_for_current_instrument(&self, price: Price) -> Option<Price> {
1395 if !self.price_matches_current_instrument(price) {
1396 return None;
1397 }
1398
1399 Some(Price::from_raw(
1400 price.raw,
1401 self.instrument.price_precision(),
1402 ))
1403 }
1404
1405 fn normalize_quantity_for_current_instrument(&self, quantity: Quantity) -> Option<Quantity> {
1406 let precision = self.instrument.size_precision();
1407 if !Self::quantity_matches_precision(quantity, precision) {
1408 return None;
1409 }
1410
1411 Some(Quantity::from_raw(quantity.raw, precision))
1412 }
1413
1414 pub fn process_order_book_delta(&mut self, delta: &OrderBookDelta) -> anyhow::Result<()> {
1422 log::debug!("Processing {delta}");
1423
1424 if matches!(delta.action, BookAction::Add | BookAction::Update) {
1426 self.check_price_precision(delta.order.price.precision, "delta order price")?;
1427 self.check_size_precision(delta.order.size.precision, "delta order size")?;
1428 }
1429
1430 if self.book_type == BookType::L1_MBP {
1432 self.iterate(delta.ts_init, AggressorSide::NoAggressor);
1433 return Ok(());
1434 }
1435
1436 self.book.apply_delta(delta)?;
1437
1438 let delta_snapshot_or_clear = (delta.flags & 32) != 0 || delta.action == BookAction::Clear;
1439
1440 if self.config.queue_position {
1441 if delta_snapshot_or_clear {
1442 self.clear_all_queue_positions();
1443 } else {
1444 self.adjust_queue_for_delta(delta);
1445 }
1446 }
1447
1448 if self.config.queue_position && delta_snapshot_or_clear {
1449 self.seed_tob_baseline();
1450 }
1451
1452 self.iterate(delta.ts_init, AggressorSide::NoAggressor);
1453 Ok(())
1454 }
1455
1456 pub fn process_order_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
1464 log::debug!("Processing {deltas}");
1465
1466 for delta in &deltas.deltas {
1468 if matches!(delta.action, BookAction::Add | BookAction::Update) {
1469 self.check_price_precision(delta.order.price.precision, "delta order price")?;
1470 self.check_size_precision(delta.order.size.precision, "delta order size")?;
1471 }
1472 }
1473
1474 if self.book_type == BookType::L1_MBP {
1476 self.iterate(deltas.ts_init, AggressorSide::NoAggressor);
1477 return Ok(());
1478 }
1479
1480 self.book.apply_deltas(deltas)?;
1481
1482 let mut has_snapshot_or_clear = false;
1483
1484 if self.config.queue_position {
1485 for delta in &deltas.deltas {
1486 if (delta.flags & 32) != 0 || delta.action == BookAction::Clear {
1487 self.clear_all_queue_positions();
1488 has_snapshot_or_clear = true;
1489 break;
1490 }
1491 self.adjust_queue_for_delta(delta);
1492 }
1493 }
1494
1495 if self.config.queue_position && has_snapshot_or_clear {
1496 self.seed_tob_baseline();
1497 }
1498
1499 self.iterate(deltas.ts_init, AggressorSide::NoAggressor);
1500 Ok(())
1501 }
1502
1503 pub fn process_order_book_depth10(&mut self, depth: &OrderBookDepth10) -> anyhow::Result<()> {
1512 log::debug!("Processing OrderBookDepth10 for {}", depth.instrument_id);
1513
1514 for order in &depth.bids {
1516 if order.side.is_none() || !order.size.is_positive() {
1517 continue;
1518 }
1519 self.check_price_precision(order.price.precision, "bid price")?;
1520 self.check_size_precision(order.size.precision, "bid size")?;
1521 }
1522
1523 for order in &depth.asks {
1524 if order.side.is_none() || !order.size.is_positive() {
1525 continue;
1526 }
1527 self.check_price_precision(order.price.precision, "ask price")?;
1528 self.check_size_precision(order.size.precision, "ask size")?;
1529 }
1530
1531 let top_bid = Self::first_valid_depth_order(&depth.bids, OrderSide::Buy);
1532 let top_ask = Self::first_valid_depth_order(&depth.asks, OrderSide::Sell);
1533
1534 if self.book_type == BookType::L1_MBP {
1537 let quote = QuoteTick::new(
1538 depth.instrument_id,
1539 Self::depth_quote_price(top_bid, self.instrument.price_precision()),
1540 Self::depth_quote_price(top_ask, self.instrument.price_precision()),
1541 Self::depth_quote_size(top_bid, self.instrument.size_precision()),
1542 Self::depth_quote_size(top_ask, self.instrument.size_precision()),
1543 depth.ts_event,
1544 depth.ts_init,
1545 );
1546 self.book.update_quote_tick("e)?;
1547 self.last_quote_bid = top_bid.map(|order| order.price);
1548 self.last_quote_ask = top_ask.map(|order| order.price);
1549 } else {
1550 self.book.apply_depth(depth)?;
1551 }
1552
1553 if self.config.queue_position {
1555 self.clear_all_queue_positions();
1556 let bid_price_raw = top_bid.map_or(0, |order| order.price.raw);
1557 let bid_size_raw = top_bid.map_or(0, |order| order.size.raw);
1558 let ask_price_raw = top_ask.map_or(0, |order| order.price.raw);
1559 let ask_size_raw = top_ask.map_or(0, |order| order.size.raw);
1560
1561 if self.tob_initialized {
1563 if bid_price_raw < self.prev_bid_price_raw {
1564 self.adjust_l1_queue_on_price_move(bid_price_raw, bid_size_raw, OrderSide::Buy);
1565 }
1566
1567 if ask_price_raw > self.prev_ask_price_raw {
1568 self.adjust_l1_queue_on_price_move(
1569 ask_price_raw,
1570 ask_size_raw,
1571 OrderSide::Sell,
1572 );
1573 }
1574 }
1575
1576 self.resolve_pending_l1_snapshots(
1577 bid_price_raw,
1578 bid_size_raw,
1579 ask_price_raw,
1580 ask_size_raw,
1581 );
1582
1583 self.prev_bid_price_raw = bid_price_raw;
1584 self.prev_bid_size_raw = bid_size_raw;
1585 self.prev_ask_price_raw = ask_price_raw;
1586 self.prev_ask_size_raw = ask_size_raw;
1587 self.tob_initialized = true;
1588 }
1589
1590 self.iterate(depth.ts_init, AggressorSide::NoAggressor);
1591 Ok(())
1592 }
1593
1594 fn first_valid_depth_order(orders: &[BookOrder], side: OrderSide) -> Option<BookOrder> {
1595 orders
1596 .iter()
1597 .copied()
1598 .find(|order| order.side == Some(side) && order.size.is_positive())
1599 }
1600
1601 fn depth_quote_price(order: Option<BookOrder>, price_precision: u8) -> Price {
1602 order.map_or_else(|| Price::zero(price_precision), |order| order.price)
1603 }
1604
1605 fn depth_quote_size(order: Option<BookOrder>, size_precision: u8) -> Quantity {
1606 order.map_or_else(|| Quantity::zero(size_precision), |order| order.size)
1607 }
1608
1609 pub fn process_quote_tick(&mut self, quote: &QuoteTick) {
1611 log::debug!("Processing {quote}");
1612
1613 if let Err(e) = self.check_price_precision(quote.bid_price.precision, "bid_price") {
1614 self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
1615 return;
1616 }
1617
1618 if let Err(e) = self.check_price_precision(quote.ask_price.precision, "ask_price") {
1619 self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
1620 return;
1621 }
1622
1623 if let Err(e) = self.check_size_precision(quote.bid_size.precision, "bid_size") {
1624 self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
1625 return;
1626 }
1627
1628 if let Err(e) = self.check_size_precision(quote.ask_size.precision, "ask_size") {
1629 self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
1630 return;
1631 }
1632
1633 self.precision_mismatch_streak = 0;
1634
1635 if self.book_type == BookType::L1_MBP {
1636 if quote.ts_event < self.book.ts_last {
1638 log::warn!(
1639 "Skipping stale quote: ts_event {} < book.ts_last {} for {}",
1640 quote.ts_event,
1641 self.book.ts_last,
1642 self.book.instrument_id,
1643 );
1644 self.iterate(quote.ts_init, AggressorSide::NoAggressor);
1645 return;
1646 }
1647
1648 if !self.update_quote_tick_or_skip(quote, "quote tick") {
1649 return;
1650 }
1651
1652 if self.config.queue_position {
1653 self.decrement_l1_queue_on_quote(
1654 quote.bid_price.raw,
1655 quote.bid_size.raw,
1656 quote.ask_price.raw,
1657 quote.ask_size.raw,
1658 );
1659 self.prev_bid_price_raw = quote.bid_price.raw;
1660 self.prev_bid_size_raw = quote.bid_size.raw;
1661 self.prev_ask_price_raw = quote.ask_price.raw;
1662 self.prev_ask_size_raw = quote.ask_size.raw;
1663 self.tob_initialized = true;
1664 }
1665 self.last_quote_bid = Some(quote.bid_price);
1666 self.last_quote_ask = Some(quote.ask_price);
1667 }
1668
1669 self.iterate(quote.ts_init, AggressorSide::NoAggressor);
1670 }
1671
1672 pub fn process_bar(&mut self, bar: &Bar) {
1681 log::debug!("Processing {bar}");
1682
1683 debug_assert!(
1684 bar.high >= bar.open
1685 && bar.high >= bar.low
1686 && bar.high >= bar.close
1687 && bar.low <= bar.open
1688 && bar.low <= bar.close,
1689 "OHLC invariant violated for {bar}"
1690 );
1691
1692 if !self.config.bar_execution || self.book_type != BookType::L1_MBP {
1694 return;
1695 }
1696
1697 let bar_type = bar.bar_type;
1698
1699 if bar_type.aggregation_source() == AggregationSource::Internal {
1701 return;
1702 }
1703
1704 if let Err(e) = self.check_price_precision(bar.open.precision, "bar open") {
1705 self.log_precision_mismatch("bar", bar.instrument_id(), &e);
1706 return;
1707 }
1708
1709 if let Err(e) = self.check_price_precision(bar.high.precision, "bar high") {
1710 self.log_precision_mismatch("bar", bar.instrument_id(), &e);
1711 return;
1712 }
1713
1714 if let Err(e) = self.check_price_precision(bar.low.precision, "bar low") {
1715 self.log_precision_mismatch("bar", bar.instrument_id(), &e);
1716 return;
1717 }
1718
1719 if let Err(e) = self.check_price_precision(bar.close.precision, "bar close") {
1720 self.log_precision_mismatch("bar", bar.instrument_id(), &e);
1721 return;
1722 }
1723
1724 if let Err(e) = self.check_size_precision(bar.volume.precision, "bar volume") {
1725 self.log_precision_mismatch("bar", bar.instrument_id(), &e);
1726 return;
1727 }
1728
1729 self.precision_mismatch_streak = 0;
1730
1731 let price_type = bar_type.spec().price_type;
1732 if price_type == PriceType::Mark {
1733 log::warn!(
1734 "Cannot process bar for {} with `PriceType::Mark`, mark price bars are not supported for bar execution",
1735 bar.instrument_id(),
1736 );
1737 return;
1738 }
1739
1740 let execution_bar_type =
1741 if let Some(execution_bar_type) = self.execution_bar_types.get(&bar.instrument_id()) {
1742 execution_bar_type.to_owned()
1743 } else {
1744 self.execution_bar_types
1745 .insert(bar.instrument_id(), bar_type);
1746 self.execution_bar_deltas
1747 .insert(bar_type, bar_type.spec().timedelta());
1748 bar_type
1749 };
1750
1751 if execution_bar_type != bar_type {
1752 let mut bar_type_timedelta = self.execution_bar_deltas.get(&bar_type).copied();
1753 if bar_type_timedelta.is_none() {
1754 bar_type_timedelta = Some(bar_type.spec().timedelta());
1755 self.execution_bar_deltas
1756 .insert(bar_type, bar_type_timedelta.unwrap());
1757 }
1758
1759 if self.execution_bar_deltas.get(&execution_bar_type).unwrap()
1760 >= &bar_type_timedelta.unwrap()
1761 {
1762 self.execution_bar_types
1763 .insert(bar_type.instrument_id(), bar_type);
1764 } else {
1765 return;
1766 }
1767 }
1768
1769 match price_type {
1770 PriceType::Last | PriceType::Mid => self.process_trade_ticks_from_bar(bar),
1771 PriceType::Bid => {
1772 self.last_bar_bid = Some(bar.to_owned());
1773 self.process_quote_ticks_from_bar();
1774 }
1775 PriceType::Ask => {
1776 self.last_bar_ask = Some(bar.to_owned());
1777 self.process_quote_ticks_from_bar();
1778 }
1779 PriceType::Mark => {
1780 unreachable!("PriceType::Mark bars return before execution bar state updates")
1781 }
1782 }
1783 }
1784
1785 fn process_trade_ticks_from_bar(&mut self, bar: &Bar) {
1786 let sizes = BarTickSizes::from_volume(bar.volume, self.instrument.size_increment());
1787
1788 let aggressor_side = if self.core.last.is_none_or(|last| bar.open > last) {
1789 AggressorSide::Buy
1790 } else {
1791 AggressorSide::Sell
1792 };
1793
1794 if self.core.last.is_none() {
1796 self.fill_at_market = true;
1797
1798 if !self.process_bar_trade_tick(
1799 bar,
1800 bar.open,
1801 sizes.open,
1802 aggressor_side,
1803 "bar open trade tick",
1804 ) {
1805 return;
1806 }
1807 self.core.set_last_raw(bar.open);
1808 } else if self.core.last.is_some_and(|last| bar.open != last) {
1809 self.fill_at_market = true;
1811
1812 if !self.process_bar_trade_tick(
1813 bar,
1814 bar.open,
1815 sizes.open,
1816 aggressor_side,
1817 "bar gap-open trade tick",
1818 ) {
1819 return;
1820 }
1821 self.core.set_last_raw(bar.open);
1822 }
1823
1824 let high_first = !self.config.bar_adaptive_high_low_ordering
1827 || (bar.high.raw - bar.open.raw).abs() < (bar.low.raw - bar.open.raw).abs();
1828
1829 if high_first {
1830 self.process_bar_high(bar, sizes.high);
1831 self.process_bar_low(bar, sizes.low);
1832 } else {
1833 self.process_bar_low(bar, sizes.low);
1834 self.process_bar_high(bar, sizes.high);
1835 }
1836
1837 if self.core.last.is_some_and(|last| bar.close != last) {
1839 self.fill_at_market = false;
1840
1841 let aggressor_side = if bar.close > self.core.last.unwrap() {
1842 AggressorSide::Buy
1843 } else {
1844 AggressorSide::Sell
1845 };
1846
1847 if !self.process_bar_trade_tick(
1848 bar,
1849 bar.close,
1850 sizes.close,
1851 aggressor_side,
1852 "bar close trade tick",
1853 ) {
1854 return;
1855 }
1856
1857 self.core.set_last_raw(bar.close);
1858 }
1859
1860 self.fill_at_market = true;
1861 }
1862
1863 fn process_bar_high(&mut self, bar: &Bar, size: Quantity) {
1864 if self.core.last.is_some_and(|last| bar.high > last) {
1865 self.fill_at_market = false;
1866
1867 if !self.process_bar_trade_tick(
1868 bar,
1869 bar.high,
1870 size,
1871 AggressorSide::Buy,
1872 "bar high trade tick",
1873 ) {
1874 return;
1875 }
1876
1877 self.core.set_last_raw(bar.high);
1878 }
1879 }
1880
1881 fn process_bar_low(&mut self, bar: &Bar, size: Quantity) {
1882 if self.core.last.is_some_and(|last| bar.low < last) {
1883 self.fill_at_market = false;
1884
1885 if !self.process_bar_trade_tick(
1886 bar,
1887 bar.low,
1888 size,
1889 AggressorSide::Sell,
1890 "bar low trade tick",
1891 ) {
1892 return;
1893 }
1894
1895 self.core.set_last_raw(bar.low);
1896 }
1897 }
1898
1899 fn process_bar_trade_tick(
1900 &mut self,
1901 bar: &Bar,
1902 price: Price,
1903 size: Quantity,
1904 aggressor_side: AggressorSide,
1905 context: &str,
1906 ) -> bool {
1907 if size.is_zero() {
1908 return true;
1909 }
1910
1911 let trade_tick = TradeTick::new(
1912 bar.instrument_id(),
1913 price,
1914 size,
1915 aggressor_side,
1916 self.ids_generator.generate_trade_id(bar.ts_init),
1917 bar.ts_init,
1918 bar.ts_init,
1919 );
1920
1921 if !self.update_trade_tick_or_skip(&trade_tick, context) {
1922 return false;
1923 }
1924
1925 self.iterate(trade_tick.ts_init, AggressorSide::NoAggressor);
1926 true
1927 }
1928
1929 fn process_quote_ticks_from_bar(&mut self) {
1930 if self.last_bar_bid.is_none()
1932 || self.last_bar_ask.is_none()
1933 || self.last_bar_bid.unwrap().ts_init != self.last_bar_ask.unwrap().ts_init
1934 {
1935 return;
1936 }
1937 let bid_bar = self.last_bar_bid.unwrap();
1938 let ask_bar = self.last_bar_ask.unwrap();
1939
1940 let size_increment = self.instrument.size_increment();
1941 let bid_sizes = BarTickSizes::from_volume(bid_bar.volume, size_increment);
1942 let ask_sizes = BarTickSizes::from_volume(ask_bar.volume, size_increment);
1943 let mut has_current_bid = false;
1944 let mut has_current_ask = false;
1945
1946 let mut quote_tick = QuoteTick::new(
1947 self.book.instrument_id,
1948 bid_bar.open,
1949 ask_bar.open,
1950 bid_sizes.open,
1951 ask_sizes.open,
1952 bid_bar.ts_init,
1953 bid_bar.ts_init,
1954 );
1955
1956 self.fill_at_market = true;
1958
1959 if !self.process_bar_quote_tick(
1960 "e_tick,
1961 "bar open quote tick",
1962 &mut has_current_bid,
1963 &mut has_current_ask,
1964 ) {
1965 return;
1966 }
1967
1968 let high_first = !self.config.bar_adaptive_high_low_ordering
1971 || (bid_bar.high.raw - bid_bar.open.raw).abs()
1972 < (bid_bar.low.raw - bid_bar.open.raw).abs();
1973
1974 let high_leg = (
1975 bid_bar.high,
1976 ask_bar.high,
1977 bid_sizes.high,
1978 ask_sizes.high,
1979 "bar high quote tick",
1980 );
1981 let low_leg = (
1982 bid_bar.low,
1983 ask_bar.low,
1984 bid_sizes.low,
1985 ask_sizes.low,
1986 "bar low quote tick",
1987 );
1988 let legs = if high_first {
1989 [high_leg, low_leg]
1990 } else {
1991 [low_leg, high_leg]
1992 };
1993
1994 for (bid_price, ask_price, bid_size, ask_size, context) in legs {
1996 self.fill_at_market = false;
1997 quote_tick.bid_price = bid_price;
1998 quote_tick.ask_price = ask_price;
1999 quote_tick.bid_size = bid_size;
2000 quote_tick.ask_size = ask_size;
2001
2002 if !self.process_bar_quote_tick(
2003 "e_tick,
2004 context,
2005 &mut has_current_bid,
2006 &mut has_current_ask,
2007 ) {
2008 return;
2009 }
2010 }
2011
2012 self.fill_at_market = false;
2014 quote_tick.bid_price = bid_bar.close;
2015 quote_tick.ask_price = ask_bar.close;
2016 quote_tick.bid_size = bid_sizes.close;
2017 quote_tick.ask_size = ask_sizes.close;
2018
2019 if !self.process_bar_quote_tick(
2020 "e_tick,
2021 "bar close quote tick",
2022 &mut has_current_bid,
2023 &mut has_current_ask,
2024 ) {
2025 return;
2026 }
2027
2028 self.last_bar_bid = None;
2029 self.last_bar_ask = None;
2030 self.fill_at_market = true;
2031 }
2032
2033 fn process_bar_quote_tick(
2034 &mut self,
2035 quote: &QuoteTick,
2036 context: &str,
2037 has_current_bid: &mut bool,
2038 has_current_ask: &mut bool,
2039 ) -> bool {
2040 let has_bid_size = !quote.bid_size.is_zero();
2041 let has_ask_size = !quote.ask_size.is_zero();
2042 let mut book_changed = false;
2043 let mut bid_cleared = false;
2044 let mut ask_cleared = false;
2045
2046 match (has_bid_size, has_ask_size) {
2047 (true, true) => {
2048 if !self.update_quote_tick_or_skip(quote, context) {
2049 return false;
2050 }
2051 *has_current_bid = true;
2052 *has_current_ask = true;
2053 book_changed = true;
2054 }
2055 _ => {
2056 if has_bid_size {
2057 self.update_bar_quote_bid(quote);
2058 *has_current_bid = true;
2059 book_changed = true;
2060 } else if !*has_current_bid {
2061 self.clear_bar_quote_bid(quote);
2062 *has_current_bid = true;
2063 book_changed = true;
2064 bid_cleared = true;
2065 }
2066
2067 if has_ask_size {
2068 self.update_bar_quote_ask(quote);
2069 *has_current_ask = true;
2070 book_changed = true;
2071 } else if !*has_current_ask {
2072 self.clear_bar_quote_ask(quote);
2073 *has_current_ask = true;
2074 book_changed = true;
2075 ask_cleared = true;
2076 }
2077 }
2078 }
2079
2080 if book_changed
2081 && let (Some(best_bid), Some(best_ask)) =
2082 (self.book.best_bid_price(), self.book.best_ask_price())
2083 && best_bid > best_ask
2084 {
2085 if has_bid_size && !has_ask_size {
2086 self.clear_bar_quote_ask(quote);
2087 ask_cleared = true;
2088 } else if has_ask_size && !has_bid_size {
2089 self.clear_bar_quote_bid(quote);
2090 bid_cleared = true;
2091 }
2092 }
2093
2094 if has_bid_size {
2095 self.last_quote_bid = Some(quote.bid_price);
2096 } else if bid_cleared {
2097 self.last_quote_bid = None;
2098 }
2099
2100 if has_ask_size {
2101 self.last_quote_ask = Some(quote.ask_price);
2102 } else if ask_cleared {
2103 self.last_quote_ask = None;
2104 }
2105
2106 if !book_changed {
2107 return true;
2108 }
2109
2110 self.iterate(quote.ts_init, AggressorSide::NoAggressor);
2111 true
2112 }
2113
2114 fn update_bar_quote_bid(&mut self, quote: &QuoteTick) {
2115 let bid = BookOrder::new(
2116 OrderSide::Buy,
2117 quote.bid_price,
2118 quote.bid_size,
2119 OrderSide::Buy as u64,
2120 );
2121 self.book
2122 .add(bid, 0, self.book.sequence.saturating_add(1), quote.ts_event);
2123 }
2124
2125 fn clear_bar_quote_bid(&mut self, quote: &QuoteTick) {
2126 self.book
2127 .clear_bids(self.book.sequence.saturating_add(1), quote.ts_event);
2128 }
2129
2130 fn update_bar_quote_ask(&mut self, quote: &QuoteTick) {
2131 let ask = BookOrder::new(
2132 OrderSide::Sell,
2133 quote.ask_price,
2134 quote.ask_size,
2135 OrderSide::Sell as u64,
2136 );
2137 self.book
2138 .add(ask, 0, self.book.sequence.saturating_add(1), quote.ts_event);
2139 }
2140
2141 fn clear_bar_quote_ask(&mut self, quote: &QuoteTick) {
2142 self.book
2143 .clear_asks(self.book.sequence.saturating_add(1), quote.ts_event);
2144 }
2145
2146 pub fn process_trade_tick(&mut self, trade: &TradeTick) {
2153 log::debug!("Processing {trade}");
2154
2155 if let Err(e) = self.check_price_precision(trade.price.precision, "trade price") {
2156 self.log_precision_mismatch("trade tick", trade.instrument_id, &e);
2157 return;
2158 }
2159
2160 if let Err(e) = self.check_size_precision(trade.size.precision, "trade size") {
2161 self.log_precision_mismatch("trade tick", trade.instrument_id, &e);
2162 return;
2163 }
2164
2165 self.precision_mismatch_streak = 0;
2166
2167 let price_raw = trade.price.raw;
2168
2169 if self.book_type == BookType::L1_MBP {
2170 if trade.ts_event < self.book.ts_last {
2172 log::warn!(
2173 "Skipping stale trade: ts_event {} < book.ts_last {} for {}",
2174 trade.ts_event,
2175 self.book.ts_last,
2176 self.book.instrument_id,
2177 );
2178 self.iterate(trade.ts_init, AggressorSide::NoAggressor);
2179 return;
2180 }
2181
2182 if !self.update_trade_tick_or_skip(trade, "trade tick") {
2183 return;
2184 }
2185 }
2186
2187 self.core.set_last_raw(trade.price);
2188
2189 if !self.config.trade_execution {
2190 if self.book_type == BookType::L1_MBP {
2192 if let Some(bid) = self.book.best_bid_price() {
2193 self.core.set_bid_raw(bid);
2194 }
2195
2196 if let Some(ask) = self.book.best_ask_price() {
2197 self.core.set_ask_raw(ask);
2198 }
2199 }
2200 return;
2201 }
2202
2203 let aggressor_side = trade.aggressor_side;
2204
2205 match aggressor_side {
2206 AggressorSide::Buy => {
2207 if self.core.ask.is_none() || price_raw > self.core.ask.map_or(0, |p| p.raw) {
2210 self.core.set_ask_raw(trade.price);
2211 }
2212
2213 if self.core.bid.is_none() {
2215 self.core.set_bid_raw(trade.price);
2216 }
2217 }
2218 AggressorSide::Sell => {
2219 if self.core.bid.is_none()
2222 || price_raw < self.core.bid.map_or(PriceRaw::MAX, |p| p.raw)
2223 {
2224 self.core.set_bid_raw(trade.price);
2225 }
2226
2227 if self.core.ask.is_none() {
2229 self.core.set_ask_raw(trade.price);
2230 }
2231 }
2232 AggressorSide::NoAggressor => {
2233 if self.core.bid.is_none()
2234 || price_raw <= self.core.bid.map_or(PriceRaw::MAX, |p| p.raw)
2235 {
2236 self.core.set_bid_raw(trade.price);
2237 }
2238
2239 if self.core.ask.is_none() || price_raw >= self.core.ask.map_or(0, |p| p.raw) {
2240 self.core.set_ask_raw(trade.price);
2241 }
2242 }
2243 }
2244
2245 let original_bid = self.core.bid;
2246 let original_ask = self.core.ask;
2247
2248 match aggressor_side {
2249 AggressorSide::Sell => {
2250 if original_ask.is_some_and(|ask| price_raw < ask.raw) {
2251 self.core.set_ask_raw(trade.price);
2252 }
2253 }
2254 AggressorSide::Buy => {
2255 if original_bid.is_some_and(|bid| price_raw > bid.raw) {
2256 self.core.set_bid_raw(trade.price);
2257 }
2258 }
2259 AggressorSide::NoAggressor => {
2260 self.core.set_bid_raw(trade.price);
2262 self.core.set_ask_raw(trade.price);
2263 }
2264 }
2265
2266 self.last_trade_size = Some(trade.size);
2267 self.trade_consumption = 0;
2268
2269 if self.config.liquidity_consumption && self.book_type != BookType::L1_MBP {
2270 self.seed_trade_consumption(price_raw, trade.size.raw, trade.ts_event, aggressor_side);
2271 }
2272
2273 self.resolve_pending_on_trade(price_raw);
2274 self.decrement_queue_on_trade(price_raw, trade.size.raw, aggressor_side);
2275
2276 self.iterate(trade.ts_init, aggressor_side);
2277
2278 self.last_trade_size = None;
2279 self.trade_consumption = 0;
2280
2281 if self.book_type == BookType::L1_MBP {
2287 match aggressor_side {
2288 AggressorSide::Sell => {
2289 if let Some(ask) = self.last_quote_ask {
2290 self.core.ask = Some(ask);
2291 }
2292 }
2293 AggressorSide::Buy => {
2294 if let Some(bid) = self.last_quote_bid {
2295 self.core.bid = Some(bid);
2296 }
2297 }
2298 AggressorSide::NoAggressor => {}
2299 }
2300 } else {
2301 match aggressor_side {
2302 AggressorSide::Sell => {
2303 if let Some(ask) = original_ask
2304 && price_raw < ask.raw
2305 {
2306 self.core.ask = Some(ask);
2307 }
2308 }
2309 AggressorSide::Buy => {
2310 if let Some(bid) = original_bid
2311 && price_raw > bid.raw
2312 {
2313 self.core.bid = Some(bid);
2314 }
2315 }
2316 AggressorSide::NoAggressor => {}
2317 }
2318 }
2319 }
2320
2321 fn update_quote_tick_or_skip(&mut self, quote: &QuoteTick, context: &str) -> bool {
2322 if let Err(e) = self.book.update_quote_tick(quote) {
2323 log::warn!(
2324 "Skipping {context} for {}: update_quote_tick failed: {e}",
2325 quote.instrument_id,
2326 );
2327 return false;
2328 }
2329 true
2330 }
2331
2332 fn update_trade_tick_or_skip(&mut self, trade: &TradeTick, context: &str) -> bool {
2333 if let Err(e) = self.book.update_trade_tick(trade) {
2334 log::warn!(
2335 "Skipping {context} for {}: update_trade_tick failed: {e}",
2336 trade.instrument_id,
2337 );
2338 return false;
2339 }
2340 true
2341 }
2342
2343 pub fn process_status(&mut self, action: MarketStatusAction) {
2345 log::debug!("Processing {action}");
2346
2347 match action {
2348 MarketStatusAction::Trading | MarketStatusAction::PreOpen
2349 if matches!(
2350 self.market_status,
2351 MarketStatus::Closed | MarketStatus::Paused | MarketStatus::Suspended
2352 ) =>
2353 {
2354 self.market_status = MarketStatus::Open;
2355 }
2356 MarketStatusAction::Pause if self.market_status == MarketStatus::Open => {
2357 self.market_status = MarketStatus::Paused;
2358 }
2359 MarketStatusAction::Suspend if self.market_status == MarketStatus::Open => {
2360 self.market_status = MarketStatus::Suspended;
2361 }
2362 MarketStatusAction::Halt | MarketStatusAction::Close
2363 if self.market_status == MarketStatus::Open =>
2364 {
2365 self.market_status = MarketStatus::Closed;
2366 }
2367 _ => {}
2368 }
2369 }
2370
2371 pub fn process_instrument_close(&mut self, close: InstrumentClose) {
2376 if close.instrument_id != self.instrument.id() {
2377 log::warn!(
2378 "Received instrument close for unknown instrument_id: {}",
2379 close.instrument_id
2380 );
2381 return;
2382 }
2383
2384 if close.close_type == InstrumentCloseType::ContractExpired {
2385 self.instrument_close = Some(close);
2386 self.iterate(close.ts_init, AggressorSide::NoAggressor);
2387 }
2388 }
2389
2390 pub fn process_instrument_expiration(&mut self, timestamp_ns: UnixNanos) {
2392 self.check_instrument_expiration(timestamp_ns);
2393 }
2394
2395 #[must_use]
2397 pub const fn is_expiration_processed(&self) -> bool {
2398 self.expiration_processed
2399 }
2400
2401 fn requires_pending_resolution(&self) -> bool {
2402 matches!(self.instrument, InstrumentAny::BinaryOption(_))
2403 }
2404
2405 fn cancel_open_orders_for_expiration(&mut self) {
2406 let instrument_id = self.instrument.id();
2411 let expiration_order_ids: IndexSet<ClientOrderId> = {
2412 let cache = self.cache.borrow();
2413 let mut order_ids = IndexSet::new();
2414
2415 for order_info in self.get_open_orders() {
2416 order_ids.insert(order_info.client_order_id);
2417 }
2418
2419 for order in cache.orders(None, Some(&instrument_id), None, None, None) {
2420 if order.is_open() || order.is_inflight() {
2421 order_ids.insert(order.client_order_id());
2422 }
2423 }
2424
2425 order_ids
2426 };
2427
2428 for client_order_id in expiration_order_ids {
2429 let order = {
2430 let cache = self.cache.borrow();
2431 cache.order(&client_order_id).map(|order| order.clone())
2432 };
2433
2434 if let Some(order) = order {
2435 self.cancel_order(&order, None);
2436 }
2437 }
2438 }
2439
2440 fn enter_pending_resolution(&mut self) {
2441 if self.pending_resolution {
2442 return;
2443 }
2444
2445 self.pending_resolution = true;
2446 self.market_status = MarketStatus::Closed;
2447 self.cancel_open_orders_for_expiration();
2448 log::info!(
2449 "{} expired and is now pending resolution; open orders canceled and new orders blocked",
2450 self.instrument.id()
2451 );
2452 }
2453
2454 fn check_instrument_expiration(&mut self, timestamp_ns: UnixNanos) {
2455 if self.expiration_processed || self.option_settlement_failed {
2456 return;
2457 }
2458
2459 let timestamp_triggered = self
2460 .instrument
2461 .expiration_ns()
2462 .is_some_and(|ns| timestamp_ns >= ns);
2463
2464 if !timestamp_triggered && self.instrument_close.is_none() {
2465 return;
2466 }
2467
2468 if self.instrument_close.is_none()
2469 && timestamp_triggered
2470 && self.requires_pending_resolution()
2471 {
2472 self.enter_pending_resolution();
2473 return;
2474 }
2475
2476 if matches!(
2477 self.instrument,
2478 InstrumentAny::OptionContract(_) | InstrumentAny::CryptoOption(_)
2479 ) {
2480 if !self.option_expiration_orders_canceled {
2484 self.option_expiration_orders_canceled = true;
2485 self.enter_pending_resolution();
2486 }
2487
2488 match self.process_option_expiry(timestamp_ns) {
2489 Ok(true) => {
2490 self.expiration_processed = true;
2491 self.pending_resolution = false;
2492 self.instrument_close.take();
2493 self.option_settlement_warning = None;
2494 log::info!("{} reached expiration", self.instrument.id());
2495 }
2496 Ok(false) => {}
2497 Err(e) => {
2498 self.option_settlement_failed = true;
2499 log::error!(
2500 "Option settlement failed terminally for {}: {e}",
2501 self.instrument.id()
2502 );
2503 }
2504 }
2505 return;
2506 }
2507
2508 self.expiration_processed = true;
2509 self.pending_resolution = false;
2510 let close = self.instrument_close.take();
2511 log::info!("{} reached expiration", self.instrument.id());
2512 self.cancel_open_orders_for_expiration();
2513
2514 let instrument_id = self.instrument.id();
2515 let positions: Vec<(
2516 TraderId,
2517 StrategyId,
2518 AccountId,
2519 PositionId,
2520 OrderSide,
2521 Quantity,
2522 )> = {
2523 let cache = self.cache.borrow();
2524 cache
2525 .positions_open(None, Some(&instrument_id), None, None, None)
2526 .into_iter()
2527 .filter_map(|pos| {
2528 OrderCore::closing_side(pos.side).map(|closing_side| {
2529 (
2530 pos.trader_id,
2531 pos.strategy_id,
2532 pos.account_id,
2533 pos.id,
2534 closing_side,
2535 pos.quantity,
2536 )
2537 })
2538 })
2539 .collect()
2540 };
2541
2542 let ts_now = self.clock.borrow().timestamp_ns();
2543 let close_price = close.as_ref().map(|close| close.close_price);
2544
2545 for (trader_id, strategy_id, account_id, position_id, closing_side, quantity) in positions {
2546 let client_order_id =
2547 ClientOrderId::from(format!("EXPIRATION-{}-{}", self.venue, UUID4::new()).as_str());
2548 let mut order = OrderAny::Market(MarketOrder::new(
2549 trader_id,
2550 strategy_id,
2551 instrument_id,
2552 client_order_id,
2553 closing_side,
2554 quantity,
2555 TimeInForce::Gtc,
2556 UUID4::new(),
2557 ts_now,
2558 true, false,
2560 None,
2561 None,
2562 None,
2563 None,
2564 None,
2565 None,
2566 None,
2567 Some(vec![Ustr::from(&format!(
2568 "EXPIRATION_{}_CLOSE",
2569 self.venue
2570 ))]),
2571 ));
2572 order.set_liquidity_side(LiquiditySide::Taker);
2573
2574 let add_result =
2575 self.cache
2576 .borrow_mut()
2577 .add_order(order.clone(), Some(position_id), None, false);
2578 if add_result.is_err() {
2579 log::debug!("Expiration order already in cache: {client_order_id}");
2580 } else {
2581 self.publish_order_initialized(&order);
2582 }
2583
2584 let venue_order_id = self.ids_generator.get_venue_order_id(&order).unwrap();
2585
2586 self.account_ids.insert(trader_id, account_id);
2589 self.generate_order_accepted(&order, venue_order_id);
2590
2591 if let Some(fill_price) = close_price {
2592 if let Err(e) = self.apply_fills(
2593 &order,
2594 &[(fill_price, quantity)],
2595 LiquiditySide::Taker,
2596 Some(position_id),
2597 None,
2598 None,
2599 ) {
2600 log::error!("Cannot fill expiration order {client_order_id}: {e}");
2601 }
2602 } else {
2603 self.fill_market_order(client_order_id);
2604 }
2605 }
2606 }
2607
2608 pub fn liquidate_open_positions(
2623 &mut self,
2624 ts_now: UnixNanos,
2625 cancel_open_orders: bool,
2626 settlement_currency: Currency,
2627 ) {
2628 if self.instrument.settlement_currency() != settlement_currency {
2630 return;
2631 }
2632
2633 if cancel_open_orders {
2634 let open_orders: Vec<RestingOrder> = self.get_open_orders();
2635 for order_info in &open_orders {
2636 let order = {
2637 let cache = self.cache.borrow();
2638 cache.order_owned(&order_info.client_order_id)
2639 };
2640
2641 if let Some(order) = order {
2642 self.cancel_order(&order, None);
2643 }
2644 }
2645 }
2646
2647 let instrument_id = self.instrument.id();
2648 let positions: Vec<(
2649 TraderId,
2650 StrategyId,
2651 AccountId,
2652 PositionId,
2653 OrderSide,
2654 Quantity,
2655 )> = {
2656 let cache = self.cache.borrow();
2657 cache
2658 .positions_open(None, Some(&instrument_id), None, None, None)
2659 .into_iter()
2660 .filter_map(|pos| {
2661 OrderCore::closing_side(pos.side).map(|closing_side| {
2662 (
2663 pos.trader_id,
2664 pos.strategy_id,
2665 pos.account_id,
2666 pos.id,
2667 closing_side,
2668 pos.quantity,
2669 )
2670 })
2671 })
2672 .collect()
2673 };
2674
2675 for (trader_id, strategy_id, account_id, position_id, closing_side, quantity) in positions {
2676 let has_price = if closing_side == OrderSide::Sell {
2678 self.best_bid_price().is_some()
2679 } else {
2680 self.best_ask_price().is_some()
2681 };
2682
2683 if !has_price {
2684 log::warn!(
2685 "LIQUIDATION: no price available for {instrument_id} position {position_id}, skipping"
2686 );
2687 continue;
2688 }
2689
2690 let client_order_id = ClientOrderId::from(
2691 format!("LIQUIDATION-{}-{}", self.venue, UUID4::new()).as_str(),
2692 );
2693 let order = OrderAny::Market(MarketOrder::new(
2694 trader_id,
2695 strategy_id,
2696 instrument_id,
2697 client_order_id,
2698 closing_side,
2699 quantity,
2700 TimeInForce::Ioc,
2701 UUID4::new(),
2702 ts_now,
2703 true, false,
2705 None,
2706 None,
2707 None,
2708 None,
2709 None,
2710 None,
2711 None,
2712 Some(vec![Ustr::from(&format!(
2713 "LIQUIDATION_{}_CLOSE",
2714 self.venue
2715 ))]),
2716 ));
2717
2718 let venue_order_id = self.ids_generator.get_venue_order_id(&order).unwrap();
2719 {
2720 let mut cache = self.cache.borrow_mut();
2721 if let Err(e) = cache.add_order(order.clone(), Some(position_id), None, false) {
2722 log::debug!("Liquidation order already in cache: {e}");
2723 } else {
2724 drop(cache);
2725 self.publish_order_initialized(&order);
2726 self.cache
2727 .borrow_mut()
2728 .add_venue_order_id(&client_order_id, &venue_order_id, false)
2729 .ok();
2730 }
2731 }
2732
2733 self.account_ids.insert(trader_id, account_id);
2736 self.generate_order_submitted(&order, account_id);
2737 self.generate_order_accepted(&order, venue_order_id);
2738 self.fill_market_order(client_order_id);
2739 }
2740 }
2741
2742 pub fn process_order(&mut self, order: &mut OrderAny, account_id: AccountId) {
2751 if self.core.order_exists(order.client_order_id()) {
2753 return;
2754 }
2755
2756 let ts_now = self.clock.borrow().timestamp_ns();
2759 self.check_instrument_expiration(ts_now);
2760
2761 let reject_reason: Option<Ustr> = 'validate: {
2766 let cache_borrow = self.cache.as_ref().borrow();
2767
2768 self.account_ids.insert(order.trader_id(), account_id);
2770
2771 if self.pending_resolution {
2772 break 'validate Some(
2773 format!(
2774 "Contract {} has expired and is pending resolution",
2775 self.instrument.id()
2776 )
2777 .into(),
2778 );
2779 }
2780
2781 if self.market_status != MarketStatus::Open {
2782 break 'validate Some(
2783 format!(
2784 "Market {} is {}, cannot accept order {}",
2785 self.instrument.id(),
2786 self.market_status,
2787 order.client_order_id()
2788 )
2789 .into(),
2790 );
2791 }
2792
2793 if self.instrument.has_expiration() {
2795 if let Some(activation_ns) = self.instrument.activation_ns()
2796 && self.clock.borrow().timestamp_ns() < activation_ns
2797 {
2798 break 'validate Some(
2799 format!(
2800 "Contract {} is not yet active, activation {activation_ns}",
2801 self.instrument.id(),
2802 )
2803 .into(),
2804 );
2805 }
2806
2807 if let Some(expiration_ns) = self.instrument.expiration_ns()
2808 && self.clock.borrow().timestamp_ns() >= expiration_ns
2809 {
2810 break 'validate Some(
2811 format!(
2812 "Contract {} has expired, expiration {expiration_ns}",
2813 self.instrument.id(),
2814 )
2815 .into(),
2816 );
2817 }
2818 }
2819
2820 if self.config.support_contingent_orders {
2822 if let Some(parent_order_id) = order.parent_order_id() {
2823 let parent_order = match cache_borrow.order(&parent_order_id) {
2824 Some(o) if o.contingency_type() == Some(ContingencyType::Oto) => o,
2825 _ => panic!("OTO parent not found"),
2826 };
2827
2828 if parent_order.status() == OrderStatus::Rejected && order.is_open() {
2829 break 'validate Some(
2830 format!("Rejected OTO order from {parent_order_id}").into(),
2831 );
2832 } else if parent_order.status() == OrderStatus::Accepted
2833 || parent_order.status() == OrderStatus::Triggered
2834 || (self.config.oto_full_trigger
2835 && parent_order.status() == OrderStatus::PartiallyFilled)
2836 {
2837 log::info!(
2838 "Pending OTO order {} triggers from {parent_order_id}",
2839 order.client_order_id(),
2840 );
2841 return;
2842 }
2843 }
2844
2845 if let Some(linked_order_ids) = order.linked_order_ids() {
2846 let contingency_type = order.contingency_type();
2847 for client_order_id in linked_order_ids {
2848 match cache_borrow.order(client_order_id) {
2849 Some(contingent_order)
2850 if matches!(
2851 contingency_type,
2852 Some(ContingencyType::Oco | ContingencyType::Ouo)
2853 ) && !order.is_closed()
2854 && contingent_order.is_closed() =>
2855 {
2856 break 'validate Some(
2857 format!("Contingent order {client_order_id} already closed")
2858 .into(),
2859 );
2860 }
2861 None => panic!("Cannot find contingent order for {client_order_id}"),
2862 _ => {}
2863 }
2864 }
2865 }
2866 }
2867
2868 if order.quantity().precision != self.instrument.size_precision() {
2870 break 'validate Some(
2871 format!(
2872 "Invalid order quantity precision for order {}, was {} when {} size precision is {}",
2873 order.client_order_id(),
2874 order.quantity().precision,
2875 self.instrument.id(),
2876 self.instrument.size_precision()
2877 )
2878 .into(),
2879 );
2880 }
2881
2882 if let Some(display_qty) = order.display_qty()
2884 && display_qty.precision != self.instrument.size_precision()
2885 {
2886 break 'validate Some(
2887 format!(
2888 "Invalid order display quantity precision for order {}, was {} when {} size precision is {}",
2889 order.client_order_id(),
2890 display_qty.precision,
2891 self.instrument.id(),
2892 self.instrument.size_precision()
2893 )
2894 .into(),
2895 );
2896 }
2897
2898 if let Some(price) = order.price()
2900 && price.precision != self.instrument.price_precision()
2901 {
2902 break 'validate Some(
2903 format!(
2904 "Invalid order price precision for order {}, was {} when {} price precision is {}",
2905 order.client_order_id(),
2906 price.precision,
2907 self.instrument.id(),
2908 self.instrument.price_precision()
2909 )
2910 .into(),
2911 );
2912 }
2913
2914 if let Some(trigger_price) = order.trigger_price()
2916 && trigger_price.precision != self.instrument.price_precision()
2917 {
2918 break 'validate Some(
2919 format!(
2920 "Invalid order trigger price precision for order {}, was {} when {} price precision is {}",
2921 order.client_order_id(),
2922 trigger_price.precision,
2923 self.instrument.id(),
2924 self.instrument.price_precision()
2925 )
2926 .into(),
2927 );
2928 }
2929
2930 let position = self.position_for_order_in_cache(&cache_borrow, order);
2931
2932 if order.order_side() == OrderSide::Sell
2934 && self.account_type != AccountType::Margin
2935 && matches!(self.instrument, InstrumentAny::Equity(_))
2936 && position
2937 .as_ref()
2938 .is_none_or(|pos| !order.would_reduce_only(pos.side, pos.quantity))
2939 {
2940 let position_string = position
2941 .as_ref()
2942 .map_or("None".to_string(), |pos| pos.id.to_string());
2943 break 'validate Some(
2944 format!(
2945 "Short selling not permitted on a CASH account with position {position_string} and order {order}",
2946 )
2947 .into(),
2948 );
2949 }
2950
2951 if self.config.use_reduce_only
2953 && order.is_reduce_only()
2954 && !order.is_closed()
2955 && position.as_ref().is_none_or(|pos| {
2956 pos.is_closed()
2957 || (order.is_buy() && pos.is_long())
2958 || (order.is_sell() && pos.is_short())
2959 })
2960 {
2961 break 'validate Some(
2962 format!(
2963 "Reduce-only order {} ({}-{}) would have increased position",
2964 order.client_order_id(),
2965 order.order_type().to_string().to_uppercase(),
2966 order.order_side().to_string().to_uppercase()
2967 )
2968 .into(),
2969 );
2970 }
2971
2972 None
2973 };
2974
2975 if let Some(reason) = reject_reason {
2976 self.generate_order_rejected(order, reason);
2977 return;
2978 }
2979
2980 if order.is_quote_quantity()
2988 && !self.instrument.is_inverse()
2989 && !matches!(
2990 order.order_type(),
2991 OrderType::TrailingStopLimit | OrderType::TrailingStopMarket,
2992 )
2993 && (order.price().is_some()
2994 || matches!(
2995 order.order_type(),
2996 OrderType::Market | OrderType::MarketToLimit,
2997 ))
2998 && !self.convert_quote_to_base_quantity(order)
2999 {
3000 return;
3001 }
3002
3003 match order.order_type() {
3004 OrderType::Market => self.process_market_order(order),
3005 OrderType::Limit => self.process_limit_order(order),
3006 OrderType::MarketToLimit => self.process_market_to_limit_order(order),
3007 OrderType::StopMarket => self.process_stop_market_order(order),
3008 OrderType::StopLimit => self.process_stop_limit_order(order),
3009 OrderType::MarketIfTouched => self.process_market_if_touched_order(order),
3010 OrderType::LimitIfTouched => self.process_limit_if_touched_order(order),
3011 OrderType::TrailingStopMarket => self.process_trailing_stop_order(order),
3012 OrderType::TrailingStopLimit => self.process_trailing_stop_order(order),
3013 }
3014 }
3015
3016 fn convert_quote_to_base_quantity(&self, order: &mut OrderAny) -> bool {
3017 let reference_price = if let Some(price) = order.price() {
3021 Some(price)
3022 } else {
3023 match order.order_side() {
3024 OrderSide::Buy => self.core.ask,
3025 OrderSide::Sell => self.core.bid,
3026 }
3027 };
3028
3029 let Some(reference_price) = reference_price else {
3030 self.generate_order_rejected(
3031 order,
3032 format!(
3033 "No market for {} to convert quote quantity to base",
3034 order.instrument_id(),
3035 )
3036 .into(),
3037 );
3038 return false;
3039 };
3040
3041 let base_quantity = self
3042 .instrument
3043 .calculate_base_quantity(order.quantity(), reference_price);
3044
3045 let ts_now = self.clock.borrow().timestamp_ns();
3046 let event = OrderEventAny::Updated(OrderUpdated::new(
3047 order.trader_id(),
3048 order.strategy_id(),
3049 order.instrument_id(),
3050 order.client_order_id(),
3051 base_quantity,
3052 UUID4::new(),
3053 ts_now,
3054 ts_now,
3055 false,
3056 order.venue_order_id(),
3057 order.account_id(),
3058 None,
3059 None,
3060 None,
3061 false,
3062 ));
3063
3064 if let Err(e) = order.apply(event.clone()) {
3068 log::error!(
3069 "Failed to apply quote-to-base update for {}: {e}",
3070 order.client_order_id(),
3071 );
3072 return false;
3073 }
3074 self.dispatch_order_event(event);
3075 true
3076 }
3077
3078 pub fn process_modify(&mut self, command: &ModifyOrder, account_id: AccountId) {
3080 if !self.core.order_exists(command.client_order_id) {
3081 self.generate_order_modify_rejected(
3082 command.trader_id,
3083 command.strategy_id,
3084 command.instrument_id,
3085 command.client_order_id,
3086 Ustr::from(format!("Order {} not found", command.client_order_id).as_str()),
3087 command.venue_order_id,
3088 Some(account_id),
3089 );
3090 return;
3091 }
3092
3093 let order = match self
3094 .cache
3095 .borrow()
3096 .order(&command.client_order_id)
3097 .map(|o| o.clone())
3098 {
3099 Some(order) => order,
3100 None => {
3101 log::error!(
3102 "Cannot modify order: order {} not found in cache",
3103 command.client_order_id
3104 );
3105 return;
3106 }
3107 };
3108
3109 let update_success = self.update_order(
3110 &order,
3111 command.quantity,
3112 command.price,
3113 command.trigger_price,
3114 None,
3115 );
3116
3117 if !update_success {
3118 return;
3119 }
3120
3121 let Some(refreshed) = self.resync_core_entry(command.client_order_id) else {
3123 return;
3124 };
3125
3126 let price_changed = refreshed.price() != order.price()
3128 || refreshed.trigger_price() != order.trigger_price();
3129
3130 if price_changed
3131 && refreshed.is_open()
3132 && self.config.queue_position
3133 && let Some(new_price) = refreshed.price()
3134 {
3135 self.snapshot_queue_position(&refreshed, new_price);
3136 self.queue_excess.swap_remove(&refreshed.client_order_id());
3137 }
3138 }
3139
3140 pub fn process_cancel(&mut self, command: &CancelOrder, account_id: AccountId) {
3142 if !self.core.order_exists(command.client_order_id) {
3143 self.generate_order_cancel_rejected(
3144 command.trader_id,
3145 command.strategy_id,
3146 account_id,
3147 command.instrument_id,
3148 command.client_order_id,
3149 command.venue_order_id,
3150 Ustr::from(format!("Order {} not found", command.client_order_id).as_str()),
3151 );
3152 return;
3153 }
3154
3155 let order = match self
3156 .cache
3157 .borrow()
3158 .order(&command.client_order_id)
3159 .map(|o| o.clone())
3160 {
3161 Some(order) => order,
3162 None => {
3163 log::error!(
3164 "Cannot cancel order: order {} not found in cache",
3165 command.client_order_id
3166 );
3167 return;
3168 }
3169 };
3170
3171 if !order.is_inflight() && !order.is_open() {
3172 self.purge_stale_core_entry(command.client_order_id);
3173 return;
3174 }
3175
3176 self.cancel_order(&order, None);
3177 }
3178
3179 pub fn process_cancel_all(&mut self, command: &CancelAllOrders, account_id: AccountId) {
3181 let instrument_id = command.instrument_id;
3182 let order_side = command.order_side;
3183
3184 let mut client_order_ids: Vec<ClientOrderId> = {
3185 let cache = self.cache.borrow();
3186 cache
3187 .orders_open_refs(
3188 None,
3189 Some(&instrument_id),
3190 None,
3191 Some(&account_id),
3192 order_side,
3193 )
3194 .into_iter()
3195 .chain(cache.orders_inflight_refs(
3196 None,
3197 Some(&instrument_id),
3198 None,
3199 Some(&account_id),
3200 order_side,
3201 ))
3202 .map(|order| order.client_order_id())
3203 .collect()
3204 };
3205 client_order_ids.sort_unstable();
3206 client_order_ids.dedup();
3207
3208 for client_order_id in client_order_ids {
3209 let order = match self
3210 .cache
3211 .borrow()
3212 .order(&client_order_id)
3213 .map(|o| o.clone())
3214 {
3215 Some(order) => order,
3216 None => continue,
3217 };
3218
3219 if !order.is_inflight() && !order.is_open() {
3220 self.purge_stale_core_entry(client_order_id);
3221 continue;
3222 }
3223
3224 self.cancel_order(&order, None);
3225 }
3226 }
3227
3228 fn purge_stale_core_entry(&mut self, client_order_id: ClientOrderId) {
3231 if self.core.order_exists(client_order_id) {
3232 self.delete_core_order(client_order_id);
3233 }
3234
3235 self.remove_queue_position(client_order_id);
3236 self.cached_filled_qty.swap_remove(&client_order_id);
3237 }
3238
3239 fn resync_core_entry(&mut self, client_order_id: ClientOrderId) -> Option<OrderAny> {
3240 let order = self
3241 .cache
3242 .borrow()
3243 .order(&client_order_id)
3244 .map(|o| o.clone())?;
3245
3246 if order.is_closed() {
3249 self.delete_core_order(client_order_id);
3250 self.remove_queue_position(client_order_id);
3251 return Some(order);
3252 }
3253
3254 let new_match_info = Self::matching_core_entry(&order);
3255
3256 let unchanged = self
3258 .core
3259 .get_order(client_order_id)
3260 .is_some_and(|existing| *existing == new_match_info);
3261
3262 if unchanged {
3263 self.track_post_match_order(&order);
3264 return Some(order);
3265 }
3266
3267 self.delete_core_order(client_order_id);
3268 self.track_post_match_order(&order);
3269 self.core.add_order(new_match_info);
3270 Some(order)
3271 }
3272
3273 pub fn process_batch_cancel(&mut self, command: &BatchCancelOrders, account_id: AccountId) {
3275 for order in &command.cancels {
3276 self.process_cancel(order, account_id);
3277 }
3278 }
3279
3280 pub fn process_batch_modify(&mut self, command: &BatchModifyOrders, account_id: AccountId) {
3282 for order in &command.modifies {
3283 self.process_modify(order, account_id);
3284 }
3285 }
3286
3287 fn process_market_order(&mut self, order: &OrderAny) {
3288 if order.time_in_force() == TimeInForce::AtTheOpen
3289 || order.time_in_force() == TimeInForce::AtTheClose
3290 {
3291 self.generate_order_rejected(
3292 order,
3293 format!(
3294 "time in force {} is not currently supported",
3295 order.time_in_force()
3296 )
3297 .into(),
3298 );
3299 return;
3300 }
3301
3302 if (order.order_side() == OrderSide::Buy && self.core.ask.is_none())
3304 || (order.order_side() == OrderSide::Sell && self.core.bid.is_none())
3305 {
3306 self.generate_order_rejected(
3307 order,
3308 format!("No market for {}", order.instrument_id()).into(),
3309 );
3310 return;
3311 }
3312
3313 if self.config.use_market_order_acks {
3314 let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
3315 self.generate_order_accepted(order, venue_order_id);
3316 }
3317
3318 if let Err(e) = self
3320 .cache
3321 .borrow_mut()
3322 .add_order(order.clone(), None, None, false)
3323 {
3324 log::debug!("Order already in cache: {e}");
3325 }
3326
3327 self.fill_market_order(order.client_order_id());
3328 }
3329
3330 fn process_limit_order(&mut self, order: &mut OrderAny) {
3331 if order.time_in_force() == TimeInForce::AtTheOpen
3332 || order.time_in_force() == TimeInForce::AtTheClose
3333 {
3334 self.generate_order_rejected(
3335 order,
3336 format!(
3337 "time in force {} is not currently supported",
3338 order.time_in_force()
3339 )
3340 .into(),
3341 );
3342 return;
3343 }
3344
3345 let limit_px = order.price().expect("Limit order must have a price");
3346 if order.is_post_only() && self.core.is_limit_matched(order.order_side(), limit_px) {
3347 self.generate_order_rejected(
3348 order,
3349 format!(
3350 "POST_ONLY {} {} order limit px of {} would have been a TAKER: bid={}, ask={}",
3351 order.order_type(),
3352 order.order_side(),
3353 order.price().unwrap(),
3354 self.core
3355 .bid
3356 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3357 self.core
3358 .ask
3359 .map_or_else(|| "None".to_string(), |p| p.to_string())
3360 )
3361 .into(),
3362 );
3363 return;
3364 }
3365
3366 self.accept_order(order);
3368
3369 if self.core.is_limit_matched(order.order_side(), limit_px) {
3371 order.set_liquidity_side(LiquiditySide::Taker);
3373
3374 if self
3375 .cache
3376 .borrow_mut()
3377 .add_order(order.clone(), None, None, false)
3378 .is_err()
3379 && let Err(e) = self.cache.borrow_mut().replace_order(order)
3380 {
3381 log::debug!("Failed to update order in cache: {e}");
3382 }
3383 self.fill_limit_order(order.client_order_id());
3384
3385 if self.core.order_exists(order.client_order_id())
3388 && let Some(mut order) = self.cache.borrow_mut().order_mut(&order.client_order_id())
3389 {
3390 order.set_liquidity_side(LiquiditySide::Maker);
3391 }
3392 } else if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc) {
3393 self.cancel_order(order, None);
3394 } else {
3395 order.set_liquidity_side(LiquiditySide::Maker);
3397
3398 if let Some(price) = order.price() {
3399 self.snapshot_queue_position(order, price);
3400 }
3401
3402 let add_result = self
3403 .cache
3404 .borrow_mut()
3405 .add_order(order.clone(), None, None, false);
3406
3407 if let Err(e) = add_result {
3408 log::debug!("Failed to add order to cache: {e}");
3409
3410 if let Some(mut order) = self.cache.borrow_mut().order_mut(&order.client_order_id())
3413 && !matches!(
3414 order.liquidity_side(),
3415 Some(LiquiditySide::Maker | LiquiditySide::Taker)
3416 )
3417 {
3418 order.set_liquidity_side(LiquiditySide::Maker);
3419 }
3420 }
3421 }
3422 }
3423
3424 fn process_market_to_limit_order(&mut self, order: &OrderAny) {
3425 if (order.order_side() == OrderSide::Buy && self.core.ask.is_none())
3427 || (order.order_side() == OrderSide::Sell && self.core.bid.is_none())
3428 {
3429 self.generate_order_rejected(
3430 order,
3431 format!("No market for {}", order.instrument_id()).into(),
3432 );
3433 return;
3434 }
3435
3436 if self.config.use_market_order_acks {
3437 let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
3438 self.generate_order_accepted(order, venue_order_id);
3439 }
3440
3441 if let Err(e) = self
3443 .cache
3444 .borrow_mut()
3445 .add_order(order.clone(), None, None, false)
3446 {
3447 log::debug!("Order already in cache: {e}");
3448 }
3449 let client_order_id = order.client_order_id();
3450 self.fill_market_order(client_order_id);
3451
3452 let filled_qty = self
3454 .cached_filled_qty
3455 .get(&client_order_id)
3456 .copied()
3457 .unwrap_or_default();
3458 let leaves_qty = order.quantity().saturating_sub(filled_qty);
3459 if leaves_qty.is_zero() {
3460 self.purge_cached_filled_qty_if_closed(client_order_id);
3461 return;
3462 }
3463
3464 let updated_order = self
3465 .cache
3466 .borrow()
3467 .order(&client_order_id)
3468 .map(|o| o.clone());
3469
3470 if let Some(mut updated_order) = updated_order {
3471 self.accept_order(&mut updated_order);
3472 }
3473 }
3474
3475 fn process_stop_market_order(&mut self, order: &mut OrderAny) {
3476 let stop_px = order
3477 .trigger_price()
3478 .expect("Stop order must have a trigger price");
3479
3480 if self.core.is_stop_matched_with_trigger_type(
3481 order.order_side(),
3482 stop_px,
3483 order.trigger_type().unwrap_or(TriggerType::Default),
3484 ) {
3485 if self.config.reject_stop_orders {
3486 self.generate_order_rejected(
3487 order,
3488 format!(
3489 "{} {} order stop px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
3490 order.order_type(),
3491 order.order_side(),
3492 order.trigger_price().unwrap(),
3493 self.core
3494 .bid
3495 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3496 self.core
3497 .ask
3498 .map_or_else(|| "None".to_string(), |p| p.to_string())
3499 ).into(),
3500 );
3501 return;
3502 }
3503
3504 if let Err(e) = self
3505 .cache
3506 .borrow_mut()
3507 .add_order(order.clone(), None, None, false)
3508 {
3509 log::debug!("Order already in cache: {e}");
3510 }
3511 self.fill_market_order(order.client_order_id());
3512 return;
3513 }
3514
3515 self.accept_order(order);
3517
3518 order.set_liquidity_side(LiquiditySide::Maker);
3520
3521 if let Err(e) = self
3522 .cache
3523 .borrow_mut()
3524 .add_order(order.clone(), None, None, false)
3525 {
3526 log::debug!("Order already in cache: {e}");
3527 }
3528 }
3529
3530 fn process_stop_limit_order(&mut self, order: &mut OrderAny) {
3531 let stop_px = order
3532 .trigger_price()
3533 .expect("Stop order must have a trigger price");
3534
3535 if self.core.is_stop_matched_with_trigger_type(
3536 order.order_side(),
3537 stop_px,
3538 order.trigger_type().unwrap_or(TriggerType::Default),
3539 ) {
3540 if self.config.reject_stop_orders {
3541 self.generate_order_rejected(
3542 order,
3543 format!(
3544 "{} {} order stop px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
3545 order.order_type(),
3546 order.order_side(),
3547 order.trigger_price().unwrap(),
3548 self.core
3549 .bid
3550 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3551 self.core
3552 .ask
3553 .map_or_else(|| "None".to_string(), |p| p.to_string())
3554 ).into(),
3555 );
3556 return;
3557 }
3558
3559 self.accept_triggered_limit_style_order(order);
3560 return;
3561 }
3562
3563 self.accept_order(order);
3564
3565 order.set_liquidity_side(LiquiditySide::Maker);
3567
3568 if let Err(e) = self
3569 .cache
3570 .borrow_mut()
3571 .add_order(order.clone(), None, None, false)
3572 {
3573 log::debug!("Order already in cache: {e}");
3574 }
3575 }
3576
3577 fn process_market_if_touched_order(&mut self, order: &mut OrderAny) {
3578 if self.core.is_touch_triggered_with_trigger_type(
3579 order.order_side(),
3580 order.trigger_price().unwrap(),
3581 order.trigger_type().unwrap_or(TriggerType::Default),
3582 ) {
3583 if self.config.reject_stop_orders {
3584 self.generate_order_rejected(
3585 order,
3586 format!(
3587 "{} {} order trigger px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
3588 order.order_type(),
3589 order.order_side(),
3590 order.trigger_price().unwrap(),
3591 self.core
3592 .bid
3593 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3594 self.core
3595 .ask
3596 .map_or_else(|| "None".to_string(), |p| p.to_string())
3597 ).into(),
3598 );
3599 return;
3600 }
3601
3602 if let Err(e) = self
3603 .cache
3604 .borrow_mut()
3605 .add_order(order.clone(), None, None, false)
3606 {
3607 log::debug!("Order already in cache: {e}");
3608 }
3609 self.fill_market_order(order.client_order_id());
3610 return;
3611 }
3612
3613 self.accept_order(order);
3615
3616 order.set_liquidity_side(LiquiditySide::Maker);
3618
3619 if let Err(e) = self
3620 .cache
3621 .borrow_mut()
3622 .add_order(order.clone(), None, None, false)
3623 {
3624 log::debug!("Order already in cache: {e}");
3625 }
3626 }
3627
3628 fn process_limit_if_touched_order(&mut self, order: &mut OrderAny) {
3629 if self.core.is_touch_triggered_with_trigger_type(
3630 order.order_side(),
3631 order.trigger_price().unwrap(),
3632 order.trigger_type().unwrap_or(TriggerType::Default),
3633 ) {
3634 if self.config.reject_stop_orders {
3635 self.generate_order_rejected(
3636 order,
3637 format!(
3638 "{} {} order trigger px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
3639 order.order_type(),
3640 order.order_side(),
3641 order.trigger_price().unwrap(),
3642 self.core
3643 .bid
3644 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3645 self.core
3646 .ask
3647 .map_or_else(|| "None".to_string(), |p| p.to_string())
3648 ).into(),
3649 );
3650 return;
3651 }
3652 self.accept_triggered_limit_style_order(order);
3653 return;
3654 }
3655
3656 self.accept_order(order);
3658
3659 order.set_liquidity_side(LiquiditySide::Maker);
3661
3662 if let Err(e) = self
3663 .cache
3664 .borrow_mut()
3665 .add_order(order.clone(), None, None, false)
3666 {
3667 log::debug!("Order already in cache: {e}");
3668 }
3669 }
3670
3671 fn accept_triggered_limit_style_order(&mut self, order: &mut OrderAny) {
3672 self.accept_order(order);
3673
3674 if let Err(e) = self
3675 .cache
3676 .borrow_mut()
3677 .add_order(order.clone(), None, None, false)
3678 {
3679 log::debug!("Order already in cache: {e}");
3680 }
3681
3682 self.trigger_limit_style_stop_order(order.client_order_id(), order.clone());
3683
3684 if let Some(cached_order) = self
3685 .cache
3686 .borrow()
3687 .order(&order.client_order_id())
3688 .map(|order| order.clone())
3689 {
3690 *order = cached_order;
3691 }
3692 }
3693
3694 fn process_trailing_stop_order(&mut self, order: &mut OrderAny) {
3695 if let Some(trigger_price) = order.trigger_price()
3696 && self.core.is_stop_matched_with_trigger_type(
3697 order.order_side(),
3698 trigger_price,
3699 order.trigger_type().unwrap_or(TriggerType::Default),
3700 )
3701 {
3702 self.generate_order_rejected(
3703 order,
3704 format!(
3705 "{} {} order trigger px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
3706 order.order_type(),
3707 order.order_side(),
3708 trigger_price,
3709 self.core
3710 .bid
3711 .map_or_else(|| "None".to_string(), |p| p.to_string()),
3712 self.core
3713 .ask
3714 .map_or_else(|| "None".to_string(), |p| p.to_string())
3715 ).into(),
3716 );
3717 return;
3718 }
3719
3720 order.set_liquidity_side(LiquiditySide::Maker);
3724
3725 self.accept_order(order);
3726
3727 if let Err(e) = self
3728 .cache
3729 .borrow_mut()
3730 .add_order(order.clone(), None, None, false)
3731 {
3732 log::debug!("Order already in cache: {e}");
3733 }
3734 }
3735
3736 pub fn iterate(&mut self, timestamp_ns: UnixNanos, aggressor_side: AggressorSide) {
3743 self.purge_closed_cached_filled_qty();
3745
3746 if aggressor_side == AggressorSide::NoAggressor && self.last_trade_size.is_none() {
3752 if self.book_type == BookType::L1_MBP {
3753 if let Some(bid) = self.book.best_bid_price() {
3754 self.core.set_bid_raw(bid);
3755 }
3756
3757 if let Some(ask) = self.book.best_ask_price() {
3758 self.core.set_ask_raw(ask);
3759 }
3760 } else {
3761 self.core.bid = self.book.best_bid_price();
3765 self.core.ask = self.book.best_ask_price();
3766 }
3767 }
3768
3769 let mut matched_order = false;
3770
3771 if self.market_status == MarketStatus::Open {
3772 for action in self.core.iterate_bids() {
3775 matched_order = true;
3776
3777 match action {
3778 MatchAction::FillLimit(id) => self.fill_limit_order(id),
3779 MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
3780 }
3781 }
3782
3783 for action in self.core.iterate_asks() {
3784 matched_order = true;
3785
3786 match action {
3787 MatchAction::FillLimit(id) => self.fill_limit_order(id),
3788 MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
3789 }
3790 }
3791 }
3792
3793 let order_ids: Vec<ClientOrderId> = if matched_order {
3794 self.core.iter_orders().map(|m| m.client_order_id).collect()
3795 } else if self.post_match_order_ids.is_empty() {
3796 Vec::new()
3797 } else {
3798 self.core
3799 .iter_orders()
3800 .filter_map(|order| {
3801 self.post_match_order_ids
3802 .contains(&order.client_order_id)
3803 .then_some(order.client_order_id)
3804 })
3805 .collect()
3806 };
3807 let support_gtd_orders = self.config.support_gtd_orders;
3808
3809 for client_order_id in order_ids {
3810 let (action, keep_tracking) = {
3811 let cache = self.cache.borrow();
3812 let Some(order) = cache.order(&client_order_id) else {
3813 self.post_match_order_ids.swap_remove(&client_order_id);
3814 continue;
3815 };
3816
3817 (
3818 post_match_order_action(&order, support_gtd_orders, timestamp_ns, |order| {
3819 order.clone()
3820 }),
3821 Self::requires_post_match_maintenance(&order),
3822 )
3823 };
3824
3825 match action {
3826 PostMatchOrderAction::RemoveClosed => {
3827 self.delete_core_order(client_order_id);
3828 self.remove_queue_position(client_order_id);
3829 self.cached_filled_qty.swap_remove(&client_order_id);
3830 continue;
3831 }
3832 PostMatchOrderAction::Expire(order) => {
3833 self.delete_core_order(client_order_id);
3834 self.cached_filled_qty.swap_remove(&client_order_id);
3835 self.expire_order(&order);
3836 continue;
3837 }
3838 PostMatchOrderAction::UpdateTrailing(mut order) => {
3839 if self.maybe_activate_trailing_stop(
3840 &mut order,
3841 self.core.bid,
3842 self.core.ask,
3843 self.core.last,
3844 ) {
3845 self.update_trailing_stop_order(&order);
3846 self.resync_core_entry(client_order_id);
3847 }
3848 }
3849 PostMatchOrderAction::NoMaintenance => {
3850 if !keep_tracking {
3851 self.post_match_order_ids.swap_remove(&client_order_id);
3852 }
3853 }
3854 }
3855
3856 if self.target_bid.is_some() || self.target_ask.is_some() || self.target_last.is_some()
3859 {
3860 if let Some(t) = self.target_bid.take() {
3861 self.core.bid = Some(t);
3862 }
3863
3864 if let Some(t) = self.target_ask.take() {
3865 self.core.ask = Some(t);
3866 }
3867
3868 if let Some(t) = self.target_last.take() {
3869 self.core.last = Some(t);
3870 }
3871 }
3872 }
3873
3874 if let Some(t) = self.target_bid.take() {
3878 self.core.bid = Some(t);
3879 }
3880
3881 if let Some(t) = self.target_ask.take() {
3882 self.core.ask = Some(t);
3883 }
3884
3885 if let Some(t) = self.target_last.take() {
3886 self.core.last = Some(t);
3887 }
3888
3889 self.core.bid = self.book.best_bid_price();
3892 self.core.ask = self.book.best_ask_price();
3893
3894 self.check_instrument_expiration(timestamp_ns);
3897 self.purge_closed_cached_filled_qty();
3898 }
3899
3900 fn get_trailing_activation_price(
3901 &self,
3902 trigger_type: TriggerType,
3903 order_side: OrderSide,
3904 bid: Option<Price>,
3905 ask: Option<Price>,
3906 last: Option<Price>,
3907 ) -> Option<Price> {
3908 match trigger_type {
3909 TriggerType::LastPrice => last,
3910 TriggerType::LastOrBidAsk => last.or(match order_side {
3911 OrderSide::Buy => ask,
3912 OrderSide::Sell => bid,
3913 }),
3914
3915 _ => match order_side {
3917 OrderSide::Buy => ask,
3918 OrderSide::Sell => bid,
3919 },
3920 }
3921 }
3922
3923 fn maybe_activate_trailing_stop(
3924 &self,
3925 order: &mut OrderAny,
3926 bid: Option<Price>,
3927 ask: Option<Price>,
3928 last: Option<Price>,
3929 ) -> bool {
3930 match order {
3931 OrderAny::TrailingStopMarket(inner) => {
3932 if inner.is_activated {
3933 return true;
3934 }
3935
3936 if inner.activation_price.is_none() {
3937 let px = self.get_trailing_activation_price(
3938 inner.trigger_type,
3939 inner.order_side(),
3940 bid,
3941 ask,
3942 last,
3943 );
3944
3945 if let Some(p) = px {
3946 inner.activation_price = Some(p);
3947 inner.set_activated();
3948
3949 if let Err(e) = self.cache.borrow_mut().replace_order(order) {
3950 log::error!("Failed to update order: {e}");
3951 }
3952 return true;
3953 }
3954 return false;
3955 }
3956
3957 let activation_price = inner.activation_price.unwrap();
3958 let hit = match inner.order_side() {
3959 OrderSide::Buy => ask.is_some_and(|a| a <= activation_price),
3960 OrderSide::Sell => bid.is_some_and(|b| b >= activation_price),
3961 };
3962
3963 if hit {
3964 inner.set_activated();
3965
3966 if let Err(e) = self.cache.borrow_mut().replace_order(order) {
3967 log::error!("Failed to update order: {e}");
3968 }
3969 }
3970 hit
3971 }
3972 OrderAny::TrailingStopLimit(inner) => {
3973 if inner.is_activated {
3974 return true;
3975 }
3976
3977 if inner.activation_price.is_none() {
3978 let px = self.get_trailing_activation_price(
3979 inner.trigger_type,
3980 inner.order_side(),
3981 bid,
3982 ask,
3983 last,
3984 );
3985
3986 if let Some(p) = px {
3987 inner.activation_price = Some(p);
3988 inner.set_activated();
3989
3990 if let Err(e) = self.cache.borrow_mut().replace_order(order) {
3991 log::error!("Failed to update order: {e}");
3992 }
3993 return true;
3994 }
3995 return false;
3996 }
3997
3998 let activation_price = inner.activation_price.unwrap();
3999 let hit = match inner.order_side() {
4000 OrderSide::Buy => ask.is_some_and(|a| a <= activation_price),
4001 OrderSide::Sell => bid.is_some_and(|b| b >= activation_price),
4002 };
4003
4004 if hit {
4005 inner.set_activated();
4006
4007 if let Err(e) = self.cache.borrow_mut().replace_order(order) {
4008 log::error!("Failed to update order: {e}");
4009 }
4010 }
4011 hit
4012 }
4013 _ => true,
4014 }
4015 }
4016
4017 fn determine_limit_price_and_volume(&mut self, order: &OrderAny) -> Vec<(Price, Quantity)> {
4018 match order.price() {
4019 Some(order_price) => {
4020 let mut fills = if self.config.liquidity_consumption {
4025 let size_prec = self.instrument.size_precision();
4026 self.book
4027 .get_all_crossed_levels(order.order_side(), order_price, size_prec)
4028 } else {
4029 let book_order =
4030 BookOrder::new(order.order_side(), order_price, order.quantity(), 1);
4031 self.book.simulate_fills(&book_order)
4032 };
4033
4034 if let Some(trade_size) = self.last_trade_size
4036 && let Some(trade_price) = self.core.last
4037 {
4038 let fills_at_trade_price = fills.iter().any(|(px, _)| *px == trade_price);
4039
4040 if !fills_at_trade_price
4041 && self.core.is_limit_matched(order.order_side(), order_price)
4042 {
4043 let leaves_qty = order.leaves_qty();
4046 let available_qty = if self.config.liquidity_consumption {
4047 let remaining = trade_size.raw.saturating_sub(self.trade_consumption);
4048 Quantity::from_raw(remaining, trade_size.precision)
4049 } else {
4050 trade_size
4051 };
4052
4053 let fill_qty = min(leaves_qty, available_qty);
4054
4055 if !fill_qty.is_zero() {
4056 log::debug!(
4057 "Trade execution fill: {} @ {} (trade_price={}, available: {}, book had {} fills)",
4058 fill_qty,
4059 order_price,
4060 trade_price,
4061 available_qty,
4062 fills.len()
4063 );
4064
4065 if self.config.liquidity_consumption {
4066 self.trade_consumption += fill_qty.raw;
4067 }
4068
4069 return vec![(order_price, fill_qty)];
4074 }
4075 }
4076 }
4077
4078 if fills.is_empty() {
4080 return fills;
4081 }
4082
4083 let book_prices: Vec<Price> = if self.config.liquidity_consumption {
4087 fills.iter().map(|(px, _)| *px).collect()
4088 } else {
4089 Vec::new()
4090 };
4091 let book_prices_ref: Option<&[Price]> = if book_prices.is_empty() {
4092 None
4093 } else {
4094 Some(&book_prices)
4095 };
4096
4097 if order
4099 .liquidity_side()
4100 .is_some_and(|liquidity_side| liquidity_side == LiquiditySide::Maker)
4101 {
4102 match order.order_side() {
4103 OrderSide::Buy => {
4104 let target_price = if order
4105 .trigger_price()
4106 .is_some_and(|trigger_price| order_price > trigger_price)
4107 {
4108 order.trigger_price().unwrap()
4109 } else {
4110 order_price
4111 };
4112
4113 for fill in &mut fills {
4114 let last_px = fill.0;
4115 if last_px < order_price {
4116 self.target_bid = self.core.bid;
4118 self.target_ask = self.core.ask;
4119 self.target_last = self.core.last;
4120 self.core.set_ask_raw(target_price);
4121 self.core.set_last_raw(target_price);
4122 fill.0 = target_price;
4123 }
4124 }
4125 }
4126 OrderSide::Sell => {
4127 let target_price = if order
4128 .trigger_price()
4129 .is_some_and(|trigger_price| order_price < trigger_price)
4130 {
4131 order.trigger_price().unwrap()
4132 } else {
4133 order_price
4134 };
4135
4136 for fill in &mut fills {
4137 let last_px = fill.0;
4138 if last_px > order_price {
4139 self.target_bid = self.core.bid;
4141 self.target_ask = self.core.ask;
4142 self.target_last = self.core.last;
4143 self.core.set_bid_raw(target_price);
4144 self.core.set_last_raw(target_price);
4145 fill.0 = target_price;
4146 }
4147 }
4148 }
4149 }
4150 }
4151
4152 self.apply_liquidity_consumption(
4153 fills,
4154 order.order_side(),
4155 order.leaves_qty(),
4156 book_prices_ref,
4157 )
4158 }
4159 None => panic!("Limit order must have a price"),
4160 }
4161 }
4162
4163 fn determine_market_price_and_volume(&self, order: &OrderAny) -> Vec<(Price, Quantity)> {
4164 let price = match order.order_side() {
4165 OrderSide::Buy => Price::max(FIXED_PRECISION),
4166 OrderSide::Sell => Price::min(FIXED_PRECISION),
4167 };
4168
4169 let mut fills = if self.config.liquidity_consumption {
4172 let size_prec = self.instrument.size_precision();
4173 self.book
4174 .get_all_crossed_levels(order.order_side(), price, size_prec)
4175 } else {
4176 let book_order = BookOrder::new(order.order_side(), price, order.quantity(), 0);
4177 self.book.simulate_fills(&book_order)
4178 };
4179
4180 if !self.fill_at_market
4183 && self.book_type == BookType::L1_MBP
4184 && !fills.is_empty()
4185 && matches!(
4186 order.order_type(),
4187 OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched
4188 )
4189 && let Some(trigger_price) = order.trigger_price()
4190 {
4191 fills[0] = (trigger_price, fills[0].1);
4192
4193 let mut remaining_qty = order.leaves_qty().raw;
4195 let mut capped_fills = Vec::with_capacity(fills.len());
4196
4197 for (price, qty) in fills {
4198 if remaining_qty == 0 {
4199 break;
4200 }
4201
4202 let capped_qty_raw = min(qty.raw, remaining_qty);
4203 if capped_qty_raw == 0 {
4204 continue;
4205 }
4206
4207 remaining_qty -= capped_qty_raw;
4208 capped_fills.push((price, Quantity::from_raw(capped_qty_raw, qty.precision)));
4209 }
4210
4211 return capped_fills;
4212 }
4213
4214 fills
4215 }
4216
4217 fn determine_market_fill_model_price_and_volume(
4218 &mut self,
4219 order: &OrderAny,
4220 ) -> anyhow::Result<(Vec<(Price, Quantity)>, bool)> {
4221 if let (Some(best_bid), Some(best_ask)) = (self.core.bid, self.core.ask)
4222 && let Some(book) = self.fill_model.get_orderbook_for_fill_simulation(
4223 &self.instrument,
4224 order,
4225 best_bid,
4226 best_ask,
4227 )?
4228 {
4229 let price = match order.order_side() {
4230 OrderSide::Buy => Price::max(FIXED_PRECISION),
4231 OrderSide::Sell => Price::min(FIXED_PRECISION),
4232 };
4233 let book_order = BookOrder::new(order.order_side(), price, order.quantity(), 0);
4234 let fills = book.simulate_fills(&book_order);
4235 if !fills.is_empty() {
4236 return Ok((fills, true));
4237 }
4238 }
4239 Ok((self.determine_market_price_and_volume(order), false))
4240 }
4241
4242 fn determine_limit_fill_model_price_and_volume(
4243 &mut self,
4244 order: &OrderAny,
4245 ) -> anyhow::Result<Vec<(Price, Quantity)>> {
4246 if let (Some(best_bid), Some(best_ask)) = (self.core.bid, self.core.ask)
4247 && let Some(book) = self.fill_model.get_orderbook_for_fill_simulation(
4248 &self.instrument,
4249 order,
4250 best_bid,
4251 best_ask,
4252 )?
4253 && let Some(limit_price) = order.price()
4254 {
4255 let book_order = BookOrder::new(order.order_side(), limit_price, order.quantity(), 0);
4256 let fills = book.simulate_fills(&book_order);
4257 if !fills.is_empty() {
4258 return Ok(fills);
4259 }
4260 }
4261 Ok(self.determine_limit_price_and_volume(order))
4262 }
4263
4264 pub fn fill_market_order(&mut self, client_order_id: ClientOrderId) {
4269 let mut order = match self
4270 .cache
4271 .borrow()
4272 .order(&client_order_id)
4273 .map(|o| o.clone())
4274 {
4275 Some(order) => order,
4276 None => {
4277 log::error!("Cannot fill market order: order {client_order_id} not found in cache");
4278 return;
4279 }
4280 };
4281
4282 if order.is_closed() {
4283 self.purge_stale_core_entry(client_order_id);
4284 return;
4285 }
4286
4287 if order.is_quote_quantity()
4291 && !self.instrument.is_inverse()
4292 && !self.convert_quote_to_base_quantity(&mut order)
4293 {
4294 return;
4295 }
4296
4297 if let Some(filled_qty) = self.cached_filled_qty.get(&order.client_order_id())
4298 && filled_qty >= &order.quantity()
4299 {
4300 log::debug!(
4301 "Ignoring fill as already filled pending application of events: {:?}, {:?}, {:?}, {:?}",
4302 filled_qty,
4303 order.quantity(),
4304 order.filled_qty(),
4305 order.quantity()
4306 );
4307 return;
4308 }
4309
4310 let (venue_position_id, position) = self.fill_position_for_order(&order, Some(true));
4311
4312 if self.config.use_reduce_only && order.is_reduce_only() && position.is_none() {
4313 log::warn!(
4314 "Canceling REDUCE_ONLY {} as would increase position",
4315 order.order_type()
4316 );
4317 self.cancel_order(&order, None);
4318 return;
4319 }
4320
4321 order.set_liquidity_side(LiquiditySide::Taker);
4322 let (mut fills, from_synthetic) =
4323 match self.determine_market_fill_model_price_and_volume(&order) {
4324 Ok(result) => result,
4325 Err(e) => {
4326 log::error!(
4327 "Cannot fill market order {}: fill model failed: {e}",
4328 order.client_order_id()
4329 );
4330 return;
4331 }
4332 };
4333
4334 let protection_price: Option<Price> = if let Some(protection_points) =
4336 self.config.price_protection_points
4337 && matches!(
4338 order.order_type(),
4339 OrderType::Market | OrderType::StopMarket
4340 ) {
4341 protection_price_calculate(
4342 self.instrument.price_increment(),
4343 &order,
4344 protection_points,
4345 self.core.bid,
4346 self.core.ask,
4347 )
4348 .ok()
4349 } else {
4350 None
4351 };
4352
4353 if let Some(protection_price) = protection_price {
4354 fills = self.filter_fills_by_protection(fills, &order, protection_price);
4355 }
4356
4357 let is_trigger_price_fill = !self.fill_at_market
4360 && self.book_type == BookType::L1_MBP
4361 && matches!(
4362 order.order_type(),
4363 OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched
4364 )
4365 && order.trigger_price().is_some();
4366
4367 if !from_synthetic && !is_trigger_price_fill {
4368 fills = self.apply_liquidity_consumption(
4369 fills,
4370 order.order_side(),
4371 order.leaves_qty(),
4372 None,
4373 );
4374 }
4375
4376 if let Err(e) = self.apply_fills(
4377 &order,
4378 &fills,
4379 LiquiditySide::Taker,
4380 if self.config.use_reduce_only && order.is_reduce_only() {
4381 venue_position_id
4382 } else {
4383 None
4384 },
4385 position.as_ref(),
4386 protection_price,
4387 ) {
4388 log::error!("Cannot fill market order {}: {e}", order.client_order_id());
4389 }
4390 }
4391
4392 fn filter_fills_by_protection(
4393 &self,
4394 fills: Vec<(Price, Quantity)>,
4395 order: &OrderAny,
4396 protection_price: Price,
4397 ) -> Vec<(Price, Quantity)> {
4398 let protection_raw = protection_price.raw;
4399 fills
4400 .into_iter()
4401 .filter(|(fill_price, _)| {
4402 match order.order_side() {
4403 OrderSide::Buy => fill_price.raw <= protection_raw,
4405
4406 OrderSide::Sell => fill_price.raw >= protection_raw,
4408 }
4409 })
4410 .collect()
4411 }
4412
4413 pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
4422 let mut order = match self
4423 .cache
4424 .borrow()
4425 .order(&client_order_id)
4426 .map(|o| o.clone())
4427 {
4428 Some(order) => order,
4429 None => {
4430 log::error!("Cannot fill limit order: order {client_order_id} not found in cache");
4431 return;
4432 }
4433 };
4434
4435 if order.is_closed() {
4436 self.purge_stale_core_entry(client_order_id);
4437 return;
4438 }
4439
4440 if order.is_quote_quantity()
4444 && !self.instrument.is_inverse()
4445 && !self.convert_quote_to_base_quantity(&mut order)
4446 {
4447 return;
4448 }
4449
4450 match order.price() {
4451 Some(order_price) => {
4452 let cached_filled_qty = self.cached_filled_qty.get(&order.client_order_id());
4453 if let Some(&qty) = cached_filled_qty
4454 && qty >= order.quantity()
4455 {
4456 log::debug!(
4457 "Ignoring fill as already filled pending application of events: {}, {}, {}, {}",
4458 qty,
4459 order.quantity(),
4460 order.filled_qty(),
4461 order.leaves_qty(),
4462 );
4463 return;
4464 }
4465
4466 if order
4468 .liquidity_side()
4469 .is_some_and(|liquidity_side| liquidity_side == LiquiditySide::Maker)
4470 {
4471 let at_limit = if self.last_trade_size.is_some() && self.core.last.is_some() {
4474 self.core.last.is_some_and(|last| last == order_price)
4475 } else if order.order_side() == OrderSide::Buy {
4476 self.core.bid.is_some_and(|bid| bid == order_price)
4477 } else {
4478 self.core.ask.is_some_and(|ask| ask == order_price)
4479 };
4480
4481 if at_limit {
4482 let is_limit_filled = match self.fill_model.is_limit_filled() {
4483 Ok(value) => value,
4484 Err(e) => {
4485 log::error!(
4486 "Cannot fill limit order {}: fill model failed: {e}",
4487 order.client_order_id()
4488 );
4489 return;
4490 }
4491 };
4492
4493 if !is_limit_filled {
4494 return; }
4496 }
4497 }
4498
4499 let queue_allowed_raw = if self.config.queue_position {
4500 match self.determine_trade_fill_qty(&order) {
4501 None | Some(0) => {
4502 if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc)
4503 {
4504 self.cancel_order(&order, None);
4505 }
4506 return;
4507 }
4508 Some(allowed) => Some(allowed),
4509 }
4510 } else {
4511 None
4512 };
4513
4514 let (venue_position_id, position) = self.fill_position_for_order(&order, None);
4515
4516 if self.config.use_reduce_only && order.is_reduce_only() && position.is_none() {
4517 log::warn!(
4518 "Canceling REDUCE_ONLY {} as would increase position",
4519 order.order_type()
4520 );
4521 self.cancel_order(&order, None);
4522 return;
4523 }
4524
4525 let tc_before = self.trade_consumption;
4526 let mut fills = match self.determine_limit_fill_model_price_and_volume(&order) {
4527 Ok(fills) => fills,
4528 Err(e) => {
4529 log::error!(
4530 "Cannot fill limit order {}: fill model failed: {e}",
4531 order.client_order_id()
4532 );
4533 return;
4534 }
4535 };
4536
4537 if let Some(allowed_raw) = queue_allowed_raw {
4538 let size_prec = self.instrument.size_precision();
4539 let mut remaining = allowed_raw;
4540 fills = fills
4541 .into_iter()
4542 .filter_map(|(price, qty)| {
4543 if remaining == 0 {
4544 return None;
4545 }
4546 let capped = qty.raw.min(remaining);
4547 remaining -= capped;
4548 Some((price, Quantity::from_raw(capped, size_prec)))
4549 })
4550 .collect();
4551
4552 let consumed: QuantityRaw = fills.iter().map(|(_, qty)| qty.raw).sum();
4554
4555 if let Some(excess) = self.queue_excess.get_mut(&order.client_order_id()) {
4556 *excess = excess.saturating_sub(consumed);
4557 }
4558 self.trade_consumption = tc_before + consumed;
4559 }
4560
4561 if fills.is_empty() && self.config.liquidity_consumption {
4565 log::debug!(
4566 "Skipping fill for {}: no liquidity available after consumption",
4567 order.client_order_id()
4568 );
4569
4570 if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc) {
4571 self.cancel_order(&order, None);
4572 }
4573
4574 return;
4575 }
4576
4577 let liquidity_side = order.liquidity_side().unwrap();
4578 if let Err(e) = self.apply_fills(
4579 &order,
4580 &fills,
4581 liquidity_side,
4582 venue_position_id,
4583 position.as_ref(),
4584 None,
4585 ) {
4586 log::error!("Cannot fill limit order {}: {e}", order.client_order_id());
4587 }
4588 }
4589 None => panic!("Limit order must have a price"),
4590 }
4591 }
4592
4593 fn fill_position_for_order(
4594 &mut self,
4595 order: &OrderAny,
4596 generate: Option<bool>,
4597 ) -> (Option<PositionId>, Option<Position>) {
4598 if self.oms_type == OmsType::Hedging
4599 && self.config.use_reduce_only
4600 && order.is_reduce_only()
4601 {
4602 let cache = self.cache.as_ref().borrow();
4603
4604 if let Some(position) = cache.position_for_order(&order.client_order_id()) {
4605 let position = position.clone_without_events();
4606 return (Some(position.id), Some(position));
4607 }
4608
4609 if let Some(position) = Self::open_position_reduced_by_order(&cache, order) {
4610 return (Some(position.id), Some(position));
4611 }
4612 }
4613
4614 let venue_position_id = self.ids_generator.get_position_id(order, generate);
4615
4616 let position = {
4617 let cache = self.cache.as_ref().borrow();
4618 venue_position_id
4619 .as_ref()
4620 .and_then(|position_id| cache.position(position_id))
4621 .map(|position| position.clone_without_events())
4622 };
4623
4624 (venue_position_id, position)
4625 }
4626
4627 fn position_for_order_in_cache(&self, cache: &Cache, order: &OrderAny) -> Option<Position> {
4628 if let Some(position) = cache.position_for_order(&order.client_order_id()) {
4629 return Some(position.clone_without_events());
4630 }
4631
4632 if self.oms_type == OmsType::Netting {
4633 let position_id = PositionId::new(
4634 format!("{}-{}", order.instrument_id(), order.strategy_id()).as_str(),
4635 );
4636 return cache
4637 .position(&position_id)
4638 .map(|position| position.clone_without_events());
4639 }
4640
4641 if self.oms_type == OmsType::Hedging
4642 && self.config.use_reduce_only
4643 && order.is_reduce_only()
4644 {
4645 return Self::open_position_reduced_by_order(cache, order);
4646 }
4647
4648 None
4649 }
4650
4651 fn open_position_reduced_by_order(cache: &Cache, order: &OrderAny) -> Option<Position> {
4652 cache
4653 .positions_open(
4654 None,
4655 Some(&order.instrument_id()),
4656 Some(&order.strategy_id()),
4657 None,
4658 None,
4659 )
4660 .into_iter()
4661 .find(|position| order.would_reduce_only(position.side, position.quantity))
4662 .map(|position| position.clone_without_events())
4663 }
4664
4665 fn apply_fills(
4666 &mut self,
4667 order: &OrderAny,
4668 fills: &[(Price, Quantity)],
4669 liquidity_side: LiquiditySide,
4670 venue_position_id: Option<PositionId>,
4671 position: Option<&Position>,
4672 protection_price: Option<Price>,
4673 ) -> anyhow::Result<()> {
4674 if order.time_in_force() == TimeInForce::Fok {
4675 let mut total_size = Quantity::zero(order.quantity().precision);
4676
4677 for &(fill_px, fill_qty) in fills {
4678 if self
4679 .normalize_price_for_current_instrument(fill_px)
4680 .is_some()
4681 && let Some(fill_qty) = self.normalize_quantity_for_current_instrument(fill_qty)
4682 {
4683 total_size = total_size.add(fill_qty);
4684 }
4685 }
4686
4687 if order.leaves_qty() > total_size {
4688 self.cancel_order(order, None);
4689 return Ok(());
4690 }
4691 }
4692
4693 if fills.is_empty() {
4694 if order.status() == OrderStatus::Submitted {
4695 self.generate_order_rejected(
4696 order,
4697 format!("No market for {}", order.instrument_id()).into(),
4698 );
4699 } else {
4700 log::error!(
4701 "Cannot fill order: no fills from book when fills were expected (check size in data)"
4702 );
4703 return Ok(());
4704 }
4705 }
4706
4707 let venue_position_id = if self.oms_type == OmsType::Netting {
4709 None
4710 } else {
4711 venue_position_id
4712 };
4713
4714 let mut initial_market_to_limit_fill = false;
4715 let mut total_filled = self
4716 .cached_filled_qty
4717 .get(&order.client_order_id())
4718 .copied()
4719 .unwrap_or_else(|| order.filled_qty());
4720 let initial_total_filled = total_filled;
4721 let mut last_fill_px: Option<Price> = None;
4722 let mut reduce_only_remaining_raw = None;
4723 let mut reduce_only_filled_raw = None;
4724
4725 if self.config.use_reduce_only
4726 && order.is_reduce_only()
4727 && let Some(current_position) = position
4728 {
4729 reduce_only_remaining_raw = Some(current_position.quantity.raw);
4730 reduce_only_filled_raw = Some(total_filled.raw);
4731 }
4732
4733 for &(fill_px, fill_qty) in fills {
4734 let Some(mut fill_px) = self.normalize_fill_price(fill_px, order.client_order_id())
4735 else {
4736 continue;
4737 };
4738
4739 let Some(fill_qty) = self.normalize_fill_quantity(fill_qty, order.client_order_id())
4740 else {
4741 continue;
4742 };
4743
4744 if order.filled_qty() == Quantity::zero(order.filled_qty().precision)
4745 && order.order_type() == OrderType::MarketToLimit
4746 {
4747 self.generate_order_updated(order, order.quantity(), Some(fill_px), None, None);
4748 initial_market_to_limit_fill = true;
4749 }
4750
4751 if self.book_type == BookType::L1_MBP && self.fill_model.is_slipped()? {
4752 fill_px = match order.order_side() {
4753 OrderSide::Buy => fill_px.add(self.instrument.price_increment()),
4754 OrderSide::Sell => fill_px.sub(self.instrument.price_increment()),
4755 }
4756 }
4757
4758 let mut effective_fill_qty = fill_qty;
4759
4760 if let Some(remaining_raw) = reduce_only_remaining_raw {
4761 if remaining_raw == 0 {
4762 return Ok(());
4763 }
4764
4765 if effective_fill_qty.raw > remaining_raw {
4766 effective_fill_qty =
4767 Quantity::from_raw(remaining_raw, effective_fill_qty.precision);
4768 }
4769 }
4770
4771 if fill_qty.is_zero() {
4772 if fills.len() == 1 && order.status() == OrderStatus::Submitted {
4773 self.generate_order_rejected(
4774 order,
4775 format!("No market for {}", order.instrument_id()).into(),
4776 );
4777 }
4778 return Ok(());
4779 }
4780
4781 let capped_fill_qty = min(
4783 effective_fill_qty,
4784 order.quantity().saturating_sub(total_filled),
4785 );
4786 let reduce_only_exhausts_position = reduce_only_remaining_raw
4787 .is_some_and(|remaining_raw| capped_fill_qty.raw >= remaining_raw);
4788
4789 if reduce_only_exhausts_position {
4790 let reduce_only_target_raw = reduce_only_filled_raw
4791 .unwrap_or(initial_total_filled.raw)
4792 .checked_add(capped_fill_qty.raw)
4793 .expect("Overflow occurred when adding reduce-only target quantity");
4794 let reduce_only_target =
4795 Quantity::from_raw(reduce_only_target_raw, order.quantity().precision);
4796
4797 if order.quantity() != reduce_only_target {
4798 self.generate_order_updated(order, reduce_only_target, None, None, None);
4799 }
4800 }
4801
4802 total_filled = total_filled.add(capped_fill_qty);
4803
4804 if let Some(remaining_raw) = reduce_only_remaining_raw.as_mut() {
4805 *remaining_raw = remaining_raw.saturating_sub(capped_fill_qty.raw);
4806 }
4807
4808 if let Some(filled_raw) = reduce_only_filled_raw.as_mut() {
4809 *filled_raw = filled_raw
4810 .checked_add(capped_fill_qty.raw)
4811 .expect("Overflow occurred when adding reduce-only filled quantity");
4812 }
4813
4814 self.fill_order(
4815 order,
4816 fill_px,
4817 effective_fill_qty,
4818 liquidity_side,
4819 venue_position_id,
4820 position,
4821 )?;
4822 last_fill_px = Some(fill_px);
4823
4824 if order.order_type() == OrderType::MarketToLimit && initial_market_to_limit_fill {
4825 return Ok(());
4827 }
4828
4829 if reduce_only_exhausts_position {
4830 self.purge_cached_filled_qty_if_closed(order.client_order_id());
4831 return Ok(());
4832 }
4833 }
4834
4835 let leaves_remaining = total_filled < order.quantity();
4836 let filled_in_loop = total_filled > initial_total_filled;
4837
4838 if order.time_in_force() == TimeInForce::Ioc && leaves_remaining {
4839 self.cancel_order(order, None);
4840 return Ok(());
4841 }
4842
4843 if leaves_remaining
4846 && (order.is_open() || filled_in_loop)
4847 && self.book_type == BookType::L1_MBP
4848 && matches!(
4849 order.order_type(),
4850 OrderType::Market
4851 | OrderType::MarketIfTouched
4852 | OrderType::StopMarket
4853 | OrderType::TrailingStopMarket
4854 )
4855 {
4856 let Some(last_fill_px) = last_fill_px else {
4858 return Ok(());
4859 };
4860
4861 let side = order.order_side();
4862 let slip_fill_px = match side {
4863 OrderSide::Buy => last_fill_px.add(self.instrument.price_increment()),
4864 OrderSide::Sell => last_fill_px.sub(self.instrument.price_increment()),
4865 };
4866
4867 if let Some(protection_price) = protection_price {
4868 let exceeds_boundary = match side {
4869 OrderSide::Buy => slip_fill_px.raw > protection_price.raw,
4870 OrderSide::Sell => slip_fill_px.raw < protection_price.raw,
4871 };
4872
4873 if exceeds_boundary {
4874 return Ok(());
4875 }
4876 }
4877
4878 let mut leaves_qty = order.quantity().saturating_sub(total_filled);
4879
4880 if let Some(remaining_raw) = reduce_only_remaining_raw {
4881 if remaining_raw == 0 {
4882 return Ok(());
4883 }
4884
4885 if leaves_qty.raw > remaining_raw {
4886 leaves_qty = Quantity::from_raw(remaining_raw, leaves_qty.precision);
4887 }
4888
4889 if leaves_qty.raw >= remaining_raw {
4890 let reduce_only_target_raw = reduce_only_filled_raw
4891 .unwrap_or(initial_total_filled.raw)
4892 .checked_add(leaves_qty.raw)
4893 .expect("Overflow occurred when adding reduce-only target quantity");
4894 let reduce_only_target =
4895 Quantity::from_raw(reduce_only_target_raw, order.quantity().precision);
4896
4897 if order.quantity() != reduce_only_target {
4898 self.generate_order_updated(order, reduce_only_target, None, None, None);
4899 }
4900 }
4901 }
4902
4903 if leaves_qty.is_zero() {
4904 return Ok(());
4905 }
4906
4907 self.fill_order(
4908 order,
4909 slip_fill_px,
4910 leaves_qty,
4911 liquidity_side,
4912 venue_position_id,
4913 position,
4914 )?;
4915 self.purge_cached_filled_qty_if_closed(order.client_order_id());
4916 }
4917
4918 Ok(())
4919 }
4920
4921 fn normalize_fill_price(
4922 &self,
4923 fill_px: Price,
4924 client_order_id: ClientOrderId,
4925 ) -> Option<Price> {
4926 let normalized = self.normalize_price_for_current_instrument(fill_px);
4927 if normalized.is_none() {
4928 log::warn!(
4929 "Skipping fill for {client_order_id}: fill price {fill_px} is not compatible \
4930 with {} price_precision={} price_increment={}",
4931 self.instrument.id(),
4932 self.instrument.price_precision(),
4933 self.instrument.price_increment()
4934 );
4935 }
4936 normalized
4937 }
4938
4939 fn normalize_fill_quantity(
4940 &self,
4941 fill_qty: Quantity,
4942 client_order_id: ClientOrderId,
4943 ) -> Option<Quantity> {
4944 let normalized = self.normalize_quantity_for_current_instrument(fill_qty);
4945 if normalized.is_none() {
4946 log::warn!(
4947 "Skipping fill for {client_order_id}: fill quantity {fill_qty} is not compatible \
4948 with {} size_precision={}",
4949 self.instrument.id(),
4950 self.instrument.size_precision()
4951 );
4952 }
4953 normalized
4954 }
4955
4956 fn fill_order(
4957 &mut self,
4958 order: &OrderAny,
4959 last_px: Price,
4960 last_qty: Quantity,
4961 liquidity_side: LiquiditySide,
4962 venue_position_id: Option<PositionId>,
4963 _position: Option<&Position>,
4964 ) -> anyhow::Result<()> {
4965 self.check_size_precision(last_qty.precision, "fill quantity")?;
4966
4967 let (last_qty, new_filled_qty) =
4968 if let Some(filled_qty) = self.cached_filled_qty.get(&order.client_order_id()) {
4969 let leaves_qty = order.quantity().saturating_sub(*filled_qty);
4970 let last_qty = min(last_qty, leaves_qty);
4971 (last_qty, *filled_qty + last_qty)
4972 } else {
4973 let last_qty = min(last_qty, order.quantity());
4974 (last_qty, last_qty)
4975 };
4976
4977 if last_qty.is_zero() {
4978 return Ok(());
4979 }
4980
4981 let fee_order;
4982 let commission_order = {
4983 let mut cloned = order.clone();
4986 write_filled_qty(&mut cloned, new_filled_qty.saturating_sub(last_qty));
4987 if order.liquidity_side() != Some(liquidity_side) {
4988 cloned.set_liquidity_side(liquidity_side);
4989 }
4990 fee_order = cloned;
4991 &fee_order
4992 };
4993
4994 let underlying_px = self.fee_underlying_price()?;
4995 let commission = self.fee_model.get_commission_with_context(
4996 commission_order,
4997 last_qty,
4998 last_px,
4999 &self.instrument,
5000 underlying_px,
5001 )?;
5002
5003 self.cached_filled_qty
5004 .insert(order.client_order_id(), new_filled_qty);
5005
5006 let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
5007 self.generate_order_filled(
5008 order,
5009 venue_order_id,
5010 venue_position_id,
5011 last_qty,
5012 last_px,
5013 self.instrument.quote_currency(),
5014 commission,
5015 liquidity_side,
5016 );
5017
5018 let post_fill_filled_qty = self
5019 .cached_filled_qty
5020 .get(&order.client_order_id())
5021 .copied()
5022 .unwrap_or(order.filled_qty());
5023 let post_fill_leaves_qty = order.quantity().saturating_sub(post_fill_filled_qty);
5024 let fully_filled = post_fill_leaves_qty.is_zero();
5025
5026 if order.is_closed() || fully_filled {
5027 if self.core.order_exists(order.client_order_id()) {
5028 self.delete_core_order(order.client_order_id());
5029 }
5030
5031 self.remove_queue_position(order.client_order_id());
5032
5033 if order.order_type() != OrderType::MarketToLimit {
5036 self.purge_cached_filled_qty_if_closed(order.client_order_id());
5037 }
5038 }
5039
5040 if !self.config.support_contingent_orders {
5041 return Ok(());
5042 }
5043
5044 if let Some(contingency_type) = order.contingency_type() {
5045 match contingency_type {
5046 ContingencyType::Oto => {
5047 if let Some(linked_orders_ids) = order.linked_order_ids() {
5048 for client_order_id in linked_orders_ids {
5049 let mut child_order = match self.cache.borrow().order(client_order_id) {
5050 Some(child_order) => child_order.clone(),
5051 None => anyhow::bail!("Order {client_order_id} not found in cache"),
5052 };
5053
5054 if child_order.is_closed() || child_order.is_active_local() {
5055 continue;
5056 }
5057
5058 if let (None, Some(position_id)) =
5060 (child_order.position_id(), order.position_id())
5061 {
5062 self.cache
5063 .borrow_mut()
5064 .add_position_id(
5065 &position_id,
5066 &self.venue,
5067 client_order_id,
5068 &child_order.strategy_id(),
5069 )
5070 .unwrap();
5071 log::debug!(
5072 "Added position id {position_id} to cache for order {client_order_id}"
5073 );
5074 }
5075
5076 if (!child_order.is_open())
5077 || (matches!(child_order.status(), OrderStatus::PendingUpdate)
5078 && child_order
5079 .previous_status()
5080 .is_some_and(|s| matches!(s, OrderStatus::Submitted)))
5081 {
5082 let account_id = order
5083 .account_id()
5084 .or_else(|| self.account_ids.get(&order.trader_id()).copied())
5085 .ok_or_else(|| {
5086 anyhow::anyhow!(
5087 "Account ID not found for trader {}",
5088 order.trader_id()
5089 )
5090 })?;
5091 self.process_order(&mut child_order, account_id);
5092 }
5093 }
5094 } else {
5095 log::error!(
5096 "OTO order {} does not have linked orders",
5097 order.client_order_id()
5098 );
5099 }
5100 }
5101 ContingencyType::Oco => {
5102 if let Some(linked_orders_ids) = order.linked_order_ids() {
5103 for client_order_id in linked_orders_ids {
5104 let child_order = match self.cache.borrow().order(client_order_id) {
5105 Some(child_order) => child_order.clone(),
5106 None => anyhow::bail!("Order {client_order_id} not found in cache"),
5107 };
5108
5109 if child_order.is_closed() || child_order.is_active_local() {
5110 continue;
5111 }
5112
5113 self.cancel_order(&child_order, None);
5114 }
5115 } else {
5116 log::error!(
5117 "OCO order {} does not have linked orders",
5118 order.client_order_id()
5119 );
5120 }
5121 }
5122 ContingencyType::Ouo => {
5123 if let Some(linked_orders_ids) = order.linked_order_ids() {
5124 for client_order_id in linked_orders_ids {
5125 let child_order = match self.cache.borrow().order(client_order_id) {
5126 Some(child_order) => child_order.clone(),
5127 None => anyhow::bail!("Order {client_order_id} not found in cache"),
5128 };
5129
5130 if child_order.is_active_local() {
5131 continue;
5132 }
5133
5134 let child_filled_qty = self
5135 .cached_filled_qty
5136 .get(&child_order.client_order_id())
5137 .copied()
5138 .unwrap_or(child_order.filled_qty());
5139
5140 if post_fill_leaves_qty.is_zero() && child_order.is_open() {
5141 self.cancel_order(&child_order, None);
5142 } else if child_order.is_open()
5143 && child_filled_qty >= post_fill_leaves_qty
5144 {
5145 self.cancel_order(&child_order, Some(false));
5146 } else if !post_fill_leaves_qty.is_zero()
5147 && post_fill_leaves_qty != child_order.leaves_qty()
5148 {
5149 let price = child_order.price();
5150 let trigger_price = child_order.trigger_price();
5151 self.update_order(
5152 &child_order,
5153 Some(post_fill_leaves_qty),
5154 price,
5155 trigger_price,
5156 Some(false),
5157 );
5158 }
5159 }
5160 } else {
5161 log::error!(
5162 "OUO order {} does not have linked orders",
5163 order.client_order_id()
5164 );
5165 }
5166 }
5167 }
5168 }
5169
5170 Ok(())
5171 }
5172
5173 fn fee_underlying_price(&self) -> CorrectnessResult<Option<Price>> {
5174 if !matches!(
5175 self.instrument,
5176 InstrumentAny::CryptoOption(_) | InstrumentAny::OptionContract(_)
5177 ) {
5178 return Ok(None);
5179 }
5180
5181 let Some(underlying) = self.instrument.underlying() else {
5182 return Ok(None);
5183 };
5184
5185 let underlying_id = InstrumentId::from(format!("{underlying}.{}", self.venue).as_str());
5186 let instrument_id = self.instrument.id();
5187 let cache = self.cache.borrow();
5188 if let Some(price) = cache
5189 .price(&underlying_id, PriceType::Last)
5190 .or_else(|| cache.price(&underlying_id, PriceType::Mark))
5191 .or_else(|| cache.price(&underlying_id, PriceType::Mid))
5192 {
5193 return Ok(Some(price));
5194 }
5195
5196 cache
5197 .option_greeks(&instrument_id)
5198 .and_then(|greeks| greeks.underlying_price)
5199 .map(|price| Price::new_checked(price, FIXED_PRECISION))
5200 .transpose()
5201 }
5202
5203 fn cached_order_is_closed(&self, client_order_id: ClientOrderId) -> bool {
5204 self.cache
5205 .borrow()
5206 .order(&client_order_id)
5207 .is_none_or(|order| order.is_closed())
5208 }
5209
5210 fn purge_cached_filled_qty_if_closed(&mut self, client_order_id: ClientOrderId) {
5211 if self.cached_order_is_closed(client_order_id) {
5212 self.cached_filled_qty.swap_remove(&client_order_id);
5213 }
5214 }
5215
5216 fn purge_closed_cached_filled_qty(&mut self) {
5217 let client_order_ids: Vec<ClientOrderId> = self.cached_filled_qty.keys().copied().collect();
5218
5219 for client_order_id in client_order_ids {
5220 self.purge_cached_filled_qty_if_closed(client_order_id);
5221 }
5222 }
5223
5224 fn update_limit_order(
5225 &mut self,
5226 order: &OrderAny,
5227 quantity: Quantity,
5228 price: Price,
5229 ) -> ModifyOutcome {
5230 if self.core.is_limit_matched(order.order_side(), price) {
5231 if order.is_post_only() {
5232 self.generate_order_modify_rejected(
5233 order.trader_id(),
5234 order.strategy_id(),
5235 order.instrument_id(),
5236 order.client_order_id(),
5237 Ustr::from(format!(
5238 "POST_ONLY {} {} order with new limit px of {} would have been a TAKER: bid={}, ask={}",
5239 order.order_type(),
5240 order.order_side(),
5241 price,
5242 self.core.bid.map_or_else(|| "None".to_string(), |p| p.to_string()),
5243 self.core.ask.map_or_else(|| "None".to_string(), |p| p.to_string())
5244 ).as_str()),
5245 order.venue_order_id(),
5246 order.account_id(),
5247 );
5248 return ModifyOutcome::Rejected;
5249 }
5250
5251 self.generate_order_updated(order, quantity, Some(price), None, None);
5252
5253 let client_order_id = order.client_order_id();
5255 if let Some(mut order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5256 order.set_liquidity_side(LiquiditySide::Taker);
5257 }
5258 self.fill_limit_order(client_order_id);
5259 return ModifyOutcome::Applied;
5260 }
5261 self.generate_order_updated(order, quantity, Some(price), None, None);
5262 ModifyOutcome::Applied
5263 }
5264
5265 fn update_stop_market_order(
5266 &self,
5267 order: &OrderAny,
5268 quantity: Quantity,
5269 trigger_price: Price,
5270 ) -> ModifyOutcome {
5271 if self.core.is_stop_matched_with_trigger_type(
5272 order.order_side(),
5273 trigger_price,
5274 order.trigger_type().unwrap_or(TriggerType::Default),
5275 ) {
5276 self.generate_order_modify_rejected(
5277 order.trader_id(),
5278 order.strategy_id(),
5279 order.instrument_id(),
5280 order.client_order_id(),
5281 Ustr::from(
5282 format!(
5283 "{} {} order new stop px of {} was in the market: bid={}, ask={}",
5284 order.order_type(),
5285 order.order_side(),
5286 trigger_price,
5287 self.core
5288 .bid
5289 .map_or_else(|| "None".to_string(), |p| p.to_string()),
5290 self.core
5291 .ask
5292 .map_or_else(|| "None".to_string(), |p| p.to_string())
5293 )
5294 .as_str(),
5295 ),
5296 order.venue_order_id(),
5297 order.account_id(),
5298 );
5299 return ModifyOutcome::Rejected;
5300 }
5301
5302 self.generate_order_updated(order, quantity, None, Some(trigger_price), None);
5303 ModifyOutcome::Applied
5304 }
5305
5306 fn update_stop_limit_order(
5307 &mut self,
5308 order: &OrderAny,
5309 quantity: Quantity,
5310 price: Price,
5311 trigger_price: Price,
5312 ) -> ModifyOutcome {
5313 if order.is_triggered().is_some_and(|t| t) {
5314 if self.core.is_limit_matched(order.order_side(), price) {
5316 if order.is_post_only() {
5317 self.generate_order_modify_rejected(
5318 order.trader_id(),
5319 order.strategy_id(),
5320 order.instrument_id(),
5321 order.client_order_id(),
5322 Ustr::from(format!(
5323 "POST_ONLY {} {} order with new limit px of {} would have been a TAKER: bid={}, ask={}",
5324 order.order_type(),
5325 order.order_side(),
5326 price,
5327 self.core.bid.map_or_else(|| "None".to_string(), |p| p.to_string()),
5328 self.core.ask.map_or_else(|| "None".to_string(), |p| p.to_string())
5329 ).as_str()),
5330 order.venue_order_id(),
5331 order.account_id(),
5332 );
5333 return ModifyOutcome::Rejected;
5334 }
5335 self.generate_order_updated(order, quantity, Some(price), None, None);
5336
5337 let client_order_id = order.client_order_id();
5339 if let Some(mut order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5340 order.set_liquidity_side(LiquiditySide::Taker);
5341 }
5342 self.fill_limit_order(client_order_id);
5343 return ModifyOutcome::Applied;
5344 }
5345 } else {
5346 if self.core.is_stop_matched_with_trigger_type(
5348 order.order_side(),
5349 trigger_price,
5350 order.trigger_type().unwrap_or(TriggerType::Default),
5351 ) {
5352 self.generate_order_modify_rejected(
5353 order.trader_id(),
5354 order.strategy_id(),
5355 order.instrument_id(),
5356 order.client_order_id(),
5357 Ustr::from(
5358 format!(
5359 "{} {} order new stop px of {} was in the market: bid={}, ask={}",
5360 order.order_type(),
5361 order.order_side(),
5362 trigger_price,
5363 self.core
5364 .bid
5365 .map_or_else(|| "None".to_string(), |p| p.to_string()),
5366 self.core
5367 .ask
5368 .map_or_else(|| "None".to_string(), |p| p.to_string())
5369 )
5370 .as_str(),
5371 ),
5372 order.venue_order_id(),
5373 order.account_id(),
5374 );
5375 return ModifyOutcome::Rejected;
5376 }
5377 }
5378
5379 self.generate_order_updated(order, quantity, Some(price), Some(trigger_price), None);
5380 ModifyOutcome::Applied
5381 }
5382
5383 fn update_market_if_touched_order(
5384 &self,
5385 order: &OrderAny,
5386 quantity: Quantity,
5387 trigger_price: Price,
5388 ) -> ModifyOutcome {
5389 if self.core.is_touch_triggered_with_trigger_type(
5390 order.order_side(),
5391 trigger_price,
5392 order.trigger_type().unwrap_or(TriggerType::Default),
5393 ) {
5394 self.generate_order_modify_rejected(
5395 order.trader_id(),
5396 order.strategy_id(),
5397 order.instrument_id(),
5398 order.client_order_id(),
5399 Ustr::from(
5400 format!(
5401 "{} {} order new trigger px of {} was in the market: bid={}, ask={}",
5402 order.order_type(),
5403 order.order_side(),
5404 trigger_price,
5405 self.core
5406 .bid
5407 .map_or_else(|| "None".to_string(), |p| p.to_string()),
5408 self.core
5409 .ask
5410 .map_or_else(|| "None".to_string(), |p| p.to_string())
5411 )
5412 .as_str(),
5413 ),
5414 order.venue_order_id(),
5415 order.account_id(),
5416 );
5417
5418 return ModifyOutcome::Rejected;
5420 }
5421
5422 self.generate_order_updated(order, quantity, None, Some(trigger_price), None);
5423 ModifyOutcome::Applied
5424 }
5425
5426 fn update_limit_if_touched_order(
5427 &mut self,
5428 order: &OrderAny,
5429 quantity: Quantity,
5430 price: Price,
5431 trigger_price: Price,
5432 ) -> ModifyOutcome {
5433 if order.is_triggered().is_some_and(|t| t) {
5434 if self.core.is_limit_matched(order.order_side(), price) {
5436 if order.is_post_only() {
5437 self.generate_order_modify_rejected(
5438 order.trader_id(),
5439 order.strategy_id(),
5440 order.instrument_id(),
5441 order.client_order_id(),
5442 Ustr::from(format!(
5443 "POST_ONLY {} {} order with new limit px of {} would have been a TAKER: bid={}, ask={}",
5444 order.order_type(),
5445 order.order_side(),
5446 price,
5447 self.core.bid.map_or_else(|| "None".to_string(), |p| p.to_string()),
5448 self.core.ask.map_or_else(|| "None".to_string(), |p| p.to_string())
5449 ).as_str()),
5450 order.venue_order_id(),
5451 order.account_id(),
5452 );
5453
5454 return ModifyOutcome::Rejected;
5456 }
5457 self.generate_order_updated(order, quantity, Some(price), None, None);
5458
5459 let client_order_id = order.client_order_id();
5461 if let Some(mut order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5462 order.set_liquidity_side(LiquiditySide::Taker);
5463 }
5464 self.fill_limit_order(client_order_id);
5465 return ModifyOutcome::Applied;
5466 }
5467 } else {
5468 if self.core.is_touch_triggered_with_trigger_type(
5470 order.order_side(),
5471 trigger_price,
5472 order.trigger_type().unwrap_or(TriggerType::Default),
5473 ) {
5474 self.generate_order_modify_rejected(
5475 order.trader_id(),
5476 order.strategy_id(),
5477 order.instrument_id(),
5478 order.client_order_id(),
5479 Ustr::from(
5480 format!(
5481 "{} {} order new trigger px of {} was in the market: bid={}, ask={}",
5482 order.order_type(),
5483 order.order_side(),
5484 trigger_price,
5485 self.core
5486 .bid
5487 .map_or_else(|| "None".to_string(), |p| p.to_string()),
5488 self.core
5489 .ask
5490 .map_or_else(|| "None".to_string(), |p| p.to_string())
5491 )
5492 .as_str(),
5493 ),
5494 order.venue_order_id(),
5495 order.account_id(),
5496 );
5497 return ModifyOutcome::Rejected;
5498 }
5499 }
5500
5501 self.generate_order_updated(order, quantity, Some(price), Some(trigger_price), None);
5502 ModifyOutcome::Applied
5503 }
5504
5505 fn update_trailing_stop_order(&self, order: &OrderAny) {
5506 let (new_trigger_price, new_price) = match trailing_stop_calculate(
5507 self.instrument.price_increment(),
5508 order.trigger_price(),
5509 order,
5510 self.core.bid,
5511 self.core.ask,
5512 self.core.last,
5513 ) {
5514 Ok(prices) => prices,
5515 Err(e) => {
5516 log::debug!("Cannot calculate trailing-stop update: {e}");
5518 return;
5519 }
5520 };
5521
5522 if new_trigger_price.is_none() && new_price.is_none() {
5523 return;
5524 }
5525
5526 self.generate_order_updated(order, order.quantity(), new_price, new_trigger_price, None);
5527 }
5528
5529 fn accept_order(&mut self, order: &mut OrderAny) {
5530 if order.is_closed() {
5531 return;
5533 }
5534
5535 if order.status() != OrderStatus::Accepted {
5536 let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
5537 let event = self.create_order_accepted(order, venue_order_id);
5538
5539 if let Err(e) = order.apply(event.clone()) {
5542 log::warn!(
5543 "Skipping local apply of accepted event for {}: {e}",
5544 order.client_order_id(),
5545 );
5546 }
5547 self.dispatch_order_event(event);
5548
5549 if matches!(
5552 order.order_type(),
5553 OrderType::TrailingStopLimit | OrderType::TrailingStopMarket
5554 ) && order.trigger_price().is_none()
5555 && self.maybe_activate_trailing_stop(
5556 order,
5557 self.core.bid,
5558 self.core.ask,
5559 self.core.last,
5560 )
5561 {
5562 self.update_trailing_stop_order(order);
5563 }
5564 }
5565
5566 let match_info = Self::matching_core_entry(order);
5567 self.track_post_match_order(order);
5568 self.core.add_order(match_info);
5569 }
5570
5571 fn track_post_match_order(&mut self, order: &OrderAny) {
5572 self.post_match_order_ids.insert(order.client_order_id());
5573 }
5574
5575 fn delete_core_order(&mut self, client_order_id: ClientOrderId) {
5576 self.post_match_order_ids.swap_remove(&client_order_id);
5577 let _ = self.core.delete_order(client_order_id);
5578 }
5579
5580 fn requires_post_match_maintenance(order: &OrderAny) -> bool {
5581 order.expire_time().is_some()
5582 || matches!(
5583 order.order_type(),
5584 OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
5585 )
5586 }
5587
5588 fn matching_core_entry(order: &OrderAny) -> RestingOrder {
5589 let triggered_limit_style = matches!(
5590 order.order_type(),
5591 OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit
5592 ) && order.is_triggered().is_some_and(|triggered| triggered);
5593
5594 RestingOrder::new_with_trigger_type(
5595 order.client_order_id(),
5596 order.order_side(),
5597 order.order_type(),
5598 Some(order.trigger_type().unwrap_or(TriggerType::Default)),
5599 if triggered_limit_style {
5600 None
5601 } else {
5602 order.trigger_price()
5603 },
5604 order.price(),
5605 match order {
5606 OrderAny::TrailingStopMarket(o) => o.is_activated,
5607 OrderAny::TrailingStopLimit(o) => o.is_activated,
5608 _ => true,
5609 },
5610 )
5611 }
5612
5613 fn expire_order(&mut self, order: &OrderAny) {
5614 self.remove_queue_position(order.client_order_id());
5615
5616 if self.config.support_contingent_orders && order.contingency_type().is_some() {
5617 self.cancel_contingent_orders(order);
5618 }
5619
5620 self.generate_order_expired(order);
5621 }
5622
5623 fn cancel_order(&mut self, order: &OrderAny, cancel_contingencies: Option<bool>) {
5624 let cancel_contingencies = cancel_contingencies.unwrap_or(true);
5625
5626 if order.is_active_local()
5627 && !matches!(
5628 (order.status(), order.order_type(), order.time_in_force()),
5629 (
5630 OrderStatus::Initialized | OrderStatus::Released,
5631 OrderType::Market,
5632 TimeInForce::Ioc | TimeInForce::Fok
5633 )
5634 )
5635 {
5636 log::error!(
5637 "Cannot cancel an order with {} from the matching engine",
5638 order.status()
5639 );
5640 return;
5641 }
5642
5643 if self.core.order_exists(order.client_order_id()) {
5645 self.delete_core_order(order.client_order_id());
5646 }
5647
5648 self.remove_queue_position(order.client_order_id());
5649 self.cached_filled_qty.swap_remove(&order.client_order_id());
5650
5651 let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
5652 self.generate_order_canceled(order, venue_order_id);
5653
5654 if self.config.support_contingent_orders
5655 && order.contingency_type().is_some()
5656 && cancel_contingencies
5657 {
5658 self.cancel_contingent_orders(order);
5659 }
5660 }
5661
5662 fn update_order(
5663 &mut self,
5664 order: &OrderAny,
5665 quantity: Option<Quantity>,
5666 price: Option<Price>,
5667 trigger_price: Option<Price>,
5668 update_contingencies: Option<bool>,
5669 ) -> bool {
5670 let update_contingencies = update_contingencies.unwrap_or(true);
5671 let quantity = quantity.unwrap_or(order.quantity());
5672
5673 let price_prec = self.instrument.price_precision();
5674 let size_prec = self.instrument.size_precision();
5675 let instrument_id = self.instrument.id();
5676
5677 if quantity.precision != size_prec {
5678 self.generate_order_modify_rejected(
5679 order.trader_id(),
5680 order.strategy_id(),
5681 order.instrument_id(),
5682 order.client_order_id(),
5683 Ustr::from(&format!(
5684 "Invalid update quantity precision {}, expected {size_prec} for {instrument_id}",
5685 quantity.precision
5686 )),
5687 order.venue_order_id(),
5688 order.account_id(),
5689 );
5690 return false;
5691 }
5692
5693 if let Some(px) = price
5694 && px.precision != price_prec
5695 {
5696 self.generate_order_modify_rejected(
5697 order.trader_id(),
5698 order.strategy_id(),
5699 order.instrument_id(),
5700 order.client_order_id(),
5701 Ustr::from(&format!(
5702 "Invalid update price precision {}, expected {price_prec} for {instrument_id}",
5703 px.precision
5704 )),
5705 order.venue_order_id(),
5706 order.account_id(),
5707 );
5708 return false;
5709 }
5710
5711 if let Some(tp) = trigger_price
5712 && tp.precision != price_prec
5713 {
5714 self.generate_order_modify_rejected(
5715 order.trader_id(),
5716 order.strategy_id(),
5717 order.instrument_id(),
5718 order.client_order_id(),
5719 Ustr::from(&format!(
5720 "Invalid update trigger_price precision {}, expected {price_prec} for {instrument_id}",
5721 tp.precision
5722 )),
5723 order.venue_order_id(),
5724 order.account_id(),
5725 );
5726 return false;
5727 }
5728
5729 let filled_qty = self
5731 .cached_filled_qty
5732 .get(&order.client_order_id())
5733 .copied()
5734 .unwrap_or(order.filled_qty());
5735 if quantity < filled_qty {
5736 self.generate_order_modify_rejected(
5737 order.trader_id(),
5738 order.strategy_id(),
5739 order.instrument_id(),
5740 order.client_order_id(),
5741 Ustr::from(&format!(
5742 "Cannot reduce order quantity {quantity} below filled quantity {filled_qty}",
5743 )),
5744 order.venue_order_id(),
5745 order.account_id(),
5746 );
5747 return false;
5748 }
5749
5750 let outcome = match order {
5751 OrderAny::Limit(_) | OrderAny::MarketToLimit(_) => {
5752 let price = price.unwrap_or(order.price().unwrap());
5753 self.update_limit_order(order, quantity, price)
5754 }
5755 OrderAny::StopMarket(_) => {
5756 let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
5757 self.update_stop_market_order(order, quantity, trigger_price)
5758 }
5759 OrderAny::StopLimit(_) => {
5760 let price = price.unwrap_or(order.price().unwrap());
5761 let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
5762 self.update_stop_limit_order(order, quantity, price, trigger_price)
5763 }
5764 OrderAny::MarketIfTouched(_) => {
5765 let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
5766 self.update_market_if_touched_order(order, quantity, trigger_price)
5767 }
5768 OrderAny::LimitIfTouched(_) => {
5769 let price = price.unwrap_or(order.price().unwrap());
5770 let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
5771 self.update_limit_if_touched_order(order, quantity, price, trigger_price)
5772 }
5773 OrderAny::TrailingStopMarket(_) => {
5774 if let Some(trigger_price) = trigger_price.or(order.trigger_price()) {
5775 self.update_market_if_touched_order(order, quantity, trigger_price)
5776 } else {
5777 self.generate_order_updated(order, quantity, None, trigger_price, None);
5778 ModifyOutcome::Applied
5779 }
5780 }
5781 OrderAny::TrailingStopLimit(_) => {
5782 match (
5783 price.or(order.price()),
5784 trigger_price.or(order.trigger_price()),
5785 ) {
5786 (Some(price), Some(trigger_price)) => {
5787 self.update_limit_if_touched_order(order, quantity, price, trigger_price)
5788 }
5789 _ => {
5790 self.generate_order_updated(order, quantity, price, trigger_price, None);
5791 ModifyOutcome::Applied
5792 }
5793 }
5794 }
5795 _ => {
5796 panic!(
5797 "Unsupported order type {} for update_order",
5798 order.order_type()
5799 );
5800 }
5801 };
5802
5803 if outcome == ModifyOutcome::Rejected {
5804 return false;
5805 }
5806
5807 let new_leaves_qty = quantity.saturating_sub(filled_qty);
5809 if new_leaves_qty.is_zero() {
5810 if self.config.support_contingent_orders
5811 && order.contingency_type().is_some()
5812 && update_contingencies
5813 {
5814 self.update_contingent_order(order, quantity);
5815 }
5816
5817 self.cancel_order(order, Some(false));
5819 return true;
5820 }
5821
5822 if self.config.support_contingent_orders
5823 && order.contingency_type().is_some()
5824 && update_contingencies
5825 {
5826 self.update_contingent_order(order, quantity);
5827 }
5828
5829 true
5830 }
5831
5832 pub fn trigger_stop_order(&mut self, client_order_id: ClientOrderId) {
5834 let order = match self
5835 .cache
5836 .borrow()
5837 .order(&client_order_id)
5838 .map(|o| o.clone())
5839 {
5840 Some(order) => order,
5841 None => {
5842 log::error!(
5843 "Cannot trigger stop order: order {client_order_id} not found in cache"
5844 );
5845 return;
5846 }
5847 };
5848
5849 if order.is_closed() {
5850 log::debug!("Cannot trigger stop order: {client_order_id} already closed");
5851 return;
5852 }
5853
5854 match order.order_type() {
5855 OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
5856 self.trigger_limit_style_stop_order(client_order_id, order);
5857 }
5858 OrderType::StopMarket | OrderType::MarketIfTouched | OrderType::TrailingStopMarket => {
5859 self.fill_market_order(client_order_id);
5860 }
5861 _ => {
5862 log::error!(
5863 "Cannot trigger stop order: invalid order type {}",
5864 order.order_type()
5865 );
5866 }
5867 }
5868 }
5869
5870 fn trigger_limit_style_stop_order(&mut self, client_order_id: ClientOrderId, order: OrderAny) {
5871 if order.is_triggered().is_some_and(|triggered| triggered) {
5872 let liquidity_side = match (order.price(), order.trigger_price()) {
5873 (Some(price), Some(trigger_price)) => Self::determine_triggered_limit_liquidity(
5874 order.order_side(),
5875 price,
5876 trigger_price,
5877 ),
5878 _ => LiquiditySide::Maker,
5879 };
5880
5881 if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id)
5882 && !matches!(
5883 cached_order.liquidity_side(),
5884 Some(LiquiditySide::Maker | LiquiditySide::Taker)
5885 )
5886 {
5887 cached_order.set_liquidity_side(liquidity_side);
5888 }
5889 self.fill_limit_order(client_order_id);
5890 return;
5891 }
5892
5893 let event = self.create_order_triggered(&order);
5894 let order = match self.cache.borrow_mut().update_order(&event) {
5895 Ok(order) => order,
5896 Err(e) => {
5897 log::debug!(
5898 "Failed to apply triggered event for {} before fill: {e}",
5899 order.client_order_id(),
5900 );
5901 order
5902 }
5903 };
5904 self.dispatch_order_event(event);
5905
5906 let trigger_price = order
5907 .trigger_price()
5908 .expect("Limit-style stop order must have a trigger price");
5909 let price = order
5910 .price()
5911 .expect("Limit-style stop order must have a price");
5912
5913 let maker_inside = match order.order_side() {
5914 OrderSide::Buy => self
5915 .core
5916 .ask
5917 .is_some_and(|ask| trigger_price > price && price > ask),
5918 OrderSide::Sell => self
5919 .core
5920 .bid
5921 .is_some_and(|bid| trigger_price < price && price < bid),
5922 };
5923
5924 if maker_inside {
5925 if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5926 cached_order.set_liquidity_side(LiquiditySide::Maker);
5927 }
5928 self.resync_core_entry(client_order_id);
5929 self.fill_limit_order(client_order_id);
5930 return;
5931 }
5932
5933 if self.core.is_limit_matched(order.order_side(), price) {
5934 if order.is_post_only() {
5935 self.delete_core_order(client_order_id);
5936 self.cached_filled_qty.swap_remove(&client_order_id);
5937 let event = self.create_order_rejected(
5938 &order,
5939 format!(
5940 "POST_ONLY {} {} order limit px of {} would have been a TAKER: bid={}, ask={}",
5941 order.order_type(),
5942 order.order_side(),
5943 price,
5944 self.core
5945 .bid
5946 .map_or_else(|| "None".to_string(), |p| p.to_string()),
5947 self.core
5948 .ask
5949 .map_or_else(|| "None".to_string(), |p| p.to_string())
5950 )
5951 .into(),
5952 );
5953
5954 if let Err(e) = self.cache.borrow_mut().update_order(&event) {
5955 log::debug!(
5956 "Failed to apply rejected event for {} after post-only trigger: {e}",
5957 order.client_order_id(),
5958 );
5959 }
5960 self.dispatch_order_event(event);
5961 return;
5962 }
5963
5964 if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5965 cached_order.set_liquidity_side(LiquiditySide::Taker);
5966 }
5967 self.resync_core_entry(client_order_id);
5968 self.fill_limit_order(client_order_id);
5969 return;
5970 }
5971
5972 if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
5973 cached_order.set_liquidity_side(Self::determine_triggered_limit_liquidity(
5974 order.order_side(),
5975 price,
5976 trigger_price,
5977 ));
5978 }
5979 self.resync_core_entry(client_order_id);
5980 }
5981
5982 fn determine_triggered_limit_liquidity(
5983 side: OrderSide,
5984 price: Price,
5985 trigger_price: Price,
5986 ) -> LiquiditySide {
5987 if (side == OrderSide::Buy && trigger_price > price)
5988 || (side == OrderSide::Sell && trigger_price < price)
5989 {
5990 LiquiditySide::Maker
5991 } else {
5992 LiquiditySide::Taker
5993 }
5994 }
5995
5996 fn update_contingent_order(&mut self, order: &OrderAny, parent_quantity: Quantity) {
5997 log::debug!(
5998 "Updating contingent orders from {}",
5999 order.client_order_id()
6000 );
6001
6002 if let Some(linked_order_ids) = order.linked_order_ids() {
6003 let parent_filled_qty = self
6004 .cached_filled_qty
6005 .get(&order.client_order_id())
6006 .copied()
6007 .unwrap_or(order.filled_qty());
6008 let parent_leaves_qty = parent_quantity.saturating_sub(parent_filled_qty);
6009
6010 for client_order_id in linked_order_ids {
6011 let child_order = match self.cache.borrow().order(client_order_id) {
6012 Some(order) => order.clone(),
6013 None => panic!("Order {client_order_id} not found in cache."),
6014 };
6015
6016 if child_order.is_active_local() {
6017 continue;
6018 }
6019
6020 let child_filled_qty = self
6021 .cached_filled_qty
6022 .get(&child_order.client_order_id())
6023 .copied()
6024 .unwrap_or(child_order.filled_qty());
6025
6026 if parent_leaves_qty.is_zero() {
6027 self.cancel_order(&child_order, Some(false));
6028 } else if child_filled_qty >= parent_leaves_qty {
6029 self.cancel_order(&child_order, Some(false));
6031 } else {
6032 let child_leaves_qty = child_order.quantity().saturating_sub(child_filled_qty);
6033 if child_leaves_qty != parent_leaves_qty {
6034 let price = child_order.price();
6035 let trigger_price = child_order.trigger_price();
6036 self.update_order(
6037 &child_order,
6038 Some(parent_leaves_qty),
6039 price,
6040 trigger_price,
6041 Some(false),
6042 );
6043 }
6044 }
6045 }
6046 }
6047 }
6048
6049 fn cancel_contingent_orders(&mut self, order: &OrderAny) {
6050 if let Some(linked_order_ids) = order.linked_order_ids() {
6051 for client_order_id in linked_order_ids {
6052 let contingent_order = match self.cache.borrow().order(client_order_id) {
6053 Some(order) => order.clone(),
6054 None => panic!("Cannot find contingent order for {client_order_id}"),
6055 };
6056
6057 if contingent_order.is_active_local() {
6058 continue;
6060 }
6061
6062 if !contingent_order.is_closed() {
6063 self.cancel_order(&contingent_order, Some(false));
6064 }
6065 }
6066 }
6067 }
6068
6069 fn generate_order_submitted(&self, order: &OrderAny, account_id: AccountId) {
6070 let ts_now = self.clock.borrow().timestamp_ns();
6071 let event = OrderEventAny::Submitted(OrderSubmitted::new(
6072 order.trader_id(),
6073 order.strategy_id(),
6074 order.instrument_id(),
6075 order.client_order_id(),
6076 account_id,
6077 UUID4::new(),
6078 ts_now,
6079 ts_now,
6080 ));
6081 self.dispatch_order_event(event);
6082 }
6083
6084 fn create_order_rejected(&self, order: &OrderAny, reason: Ustr) -> OrderEventAny {
6085 let ts_now = self.clock.borrow().timestamp_ns();
6086 let account_id = order
6087 .account_id()
6088 .unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
6089
6090 let due_post_only = reason.as_str().starts_with("POST_ONLY");
6091
6092 OrderEventAny::Rejected(OrderRejected::new(
6093 order.trader_id(),
6094 order.strategy_id(),
6095 order.instrument_id(),
6096 order.client_order_id(),
6097 account_id,
6098 reason,
6099 UUID4::new(),
6100 ts_now,
6101 ts_now,
6102 false,
6103 due_post_only,
6104 ))
6105 }
6106
6107 fn generate_order_rejected(&self, order: &OrderAny, reason: Ustr) {
6108 let event = self.create_order_rejected(order, reason);
6109 self.dispatch_order_event(event);
6110 }
6111
6112 fn publish_order_initialized(&self, order: &OrderAny) {
6113 let event = OrderEventAny::Initialized(order.init_event().clone());
6114 msgbus::publish_order_event(
6115 format!("events.order.{}", order.strategy_id()).into(),
6116 &event,
6117 );
6118 }
6119
6120 fn create_order_accepted(
6121 &self,
6122 order: &OrderAny,
6123 venue_order_id: VenueOrderId,
6124 ) -> OrderEventAny {
6125 let ts_now = self.clock.borrow().timestamp_ns();
6126 let account_id = order
6127 .account_id()
6128 .unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
6129 OrderEventAny::Accepted(OrderAccepted::new(
6130 order.trader_id(),
6131 order.strategy_id(),
6132 order.instrument_id(),
6133 order.client_order_id(),
6134 venue_order_id,
6135 account_id,
6136 UUID4::new(),
6137 ts_now,
6138 ts_now,
6139 false,
6140 ))
6141 }
6142
6143 fn generate_order_accepted(&self, order: &OrderAny, venue_order_id: VenueOrderId) {
6144 let event = self.create_order_accepted(order, venue_order_id);
6145 self.dispatch_order_event(event);
6146 }
6147
6148 #[expect(clippy::too_many_arguments)]
6149 fn generate_order_modify_rejected(
6150 &self,
6151 trader_id: TraderId,
6152 strategy_id: StrategyId,
6153 instrument_id: InstrumentId,
6154 client_order_id: ClientOrderId,
6155 reason: Ustr,
6156 venue_order_id: Option<VenueOrderId>,
6157 account_id: Option<AccountId>,
6158 ) {
6159 let ts_now = self.clock.borrow().timestamp_ns();
6160 let event = OrderEventAny::ModifyRejected(OrderModifyRejected::new(
6161 trader_id,
6162 strategy_id,
6163 instrument_id,
6164 client_order_id,
6165 reason,
6166 UUID4::new(),
6167 ts_now,
6168 ts_now,
6169 false,
6170 venue_order_id,
6171 account_id,
6172 ));
6173 self.dispatch_order_event(event);
6174 }
6175
6176 #[expect(clippy::too_many_arguments)]
6177 fn generate_order_cancel_rejected(
6178 &self,
6179 trader_id: TraderId,
6180 strategy_id: StrategyId,
6181 account_id: AccountId,
6182 instrument_id: InstrumentId,
6183 client_order_id: ClientOrderId,
6184 venue_order_id: Option<VenueOrderId>,
6185 reason: Ustr,
6186 ) {
6187 let ts_now = self.clock.borrow().timestamp_ns();
6188 let event = OrderEventAny::CancelRejected(OrderCancelRejected::new(
6189 trader_id,
6190 strategy_id,
6191 instrument_id,
6192 client_order_id,
6193 reason,
6194 UUID4::new(),
6195 ts_now,
6196 ts_now,
6197 false,
6198 venue_order_id,
6199 Some(account_id),
6200 ));
6201 self.dispatch_order_event(event);
6202 }
6203
6204 fn generate_order_updated(
6205 &self,
6206 order: &OrderAny,
6207 quantity: Quantity,
6208 price: Option<Price>,
6209 trigger_price: Option<Price>,
6210 protection_price: Option<Price>,
6211 ) {
6212 let ts_now = self.clock.borrow().timestamp_ns();
6213 let event = OrderEventAny::Updated(OrderUpdated::new(
6214 order.trader_id(),
6215 order.strategy_id(),
6216 order.instrument_id(),
6217 order.client_order_id(),
6218 quantity,
6219 UUID4::new(),
6220 ts_now,
6221 ts_now,
6222 false,
6223 order.venue_order_id(),
6224 order.account_id(),
6225 price,
6226 trigger_price,
6227 protection_price,
6228 order.is_quote_quantity(),
6229 ));
6230
6231 self.dispatch_order_event(event);
6232 }
6233
6234 fn generate_order_canceled(&self, order: &OrderAny, venue_order_id: VenueOrderId) {
6235 let ts_now = self.clock.borrow().timestamp_ns();
6236 let event = OrderEventAny::Canceled(OrderCanceled::new(
6237 order.trader_id(),
6238 order.strategy_id(),
6239 order.instrument_id(),
6240 order.client_order_id(),
6241 UUID4::new(),
6242 ts_now,
6243 ts_now,
6244 false,
6245 Some(venue_order_id),
6246 order.account_id(),
6247 ));
6248 self.dispatch_order_event(event);
6249 }
6250
6251 fn create_order_triggered(&self, order: &OrderAny) -> OrderEventAny {
6252 let ts_now = self.clock.borrow().timestamp_ns();
6253 OrderEventAny::Triggered(OrderTriggered::new(
6254 order.trader_id(),
6255 order.strategy_id(),
6256 order.instrument_id(),
6257 order.client_order_id(),
6258 UUID4::new(),
6259 ts_now,
6260 ts_now,
6261 false,
6262 order.venue_order_id(),
6263 order.account_id(),
6264 ))
6265 }
6266
6267 fn generate_order_expired(&self, order: &OrderAny) {
6268 let ts_now = self.clock.borrow().timestamp_ns();
6269 let event = OrderEventAny::Expired(OrderExpired::new(
6270 order.trader_id(),
6271 order.strategy_id(),
6272 order.instrument_id(),
6273 order.client_order_id(),
6274 UUID4::new(),
6275 ts_now,
6276 ts_now,
6277 false,
6278 order.venue_order_id(),
6279 order.account_id(),
6280 ));
6281 self.dispatch_order_event(event);
6282 }
6283
6284 #[expect(clippy::too_many_arguments)]
6285 fn generate_order_filled(
6286 &mut self,
6287 order: &OrderAny,
6288 venue_order_id: VenueOrderId,
6289 venue_position_id: Option<PositionId>,
6290 last_qty: Quantity,
6291 last_px: Price,
6292 quote_currency: Currency,
6293 commission: Money,
6294 liquidity_side: LiquiditySide,
6295 ) {
6296 debug_assert!(
6297 last_qty <= order.quantity(),
6298 "Fill quantity {last_qty} exceeds order quantity {order_qty} for {client_order_id}",
6299 order_qty = order.quantity(),
6300 client_order_id = order.client_order_id()
6301 );
6302
6303 let ts_now = self.clock.borrow().timestamp_ns();
6304 let account_id = order
6305 .account_id()
6306 .unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
6307 let event = OrderEventAny::Filled(OrderFilled::new(
6308 order.trader_id(),
6309 order.strategy_id(),
6310 order.instrument_id(),
6311 order.client_order_id(),
6312 venue_order_id,
6313 account_id,
6314 self.ids_generator.generate_trade_id(ts_now),
6315 order.order_side(),
6316 order.order_type(),
6317 last_qty,
6318 last_px,
6319 quote_currency,
6320 liquidity_side,
6321 UUID4::new(),
6322 ts_now,
6323 ts_now,
6324 false,
6325 venue_position_id,
6326 Some(commission),
6327 None,
6328 ));
6329
6330 self.dispatch_order_event(event);
6331 }
6332}
6333
6334#[derive(Debug, Clone, Copy, PartialEq, Eq)]
6335enum ModifyOutcome {
6336 Applied,
6337 Rejected,
6338}
6339
6340#[derive(Debug)]
6341enum PostMatchOrderAction {
6342 RemoveClosed,
6343 Expire(OrderAny),
6344 UpdateTrailing(OrderAny),
6345 NoMaintenance,
6346}
6347
6348fn post_match_order_action<F>(
6349 order: &OrderAny,
6350 support_gtd_orders: bool,
6351 timestamp_ns: UnixNanos,
6352 clone_order: F,
6353) -> PostMatchOrderAction
6354where
6355 F: FnOnce(&OrderAny) -> OrderAny,
6356{
6357 if order.is_closed() {
6358 PostMatchOrderAction::RemoveClosed
6359 } else if support_gtd_orders
6360 && order
6361 .expire_time()
6362 .is_some_and(|expire_ns| timestamp_ns >= expire_ns)
6363 {
6364 PostMatchOrderAction::Expire(clone_order(order))
6365 } else if matches!(
6366 order.order_type(),
6367 OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
6368 ) {
6369 PostMatchOrderAction::UpdateTrailing(clone_order(order))
6370 } else {
6371 PostMatchOrderAction::NoMaintenance
6372 }
6373}
6374
6375fn write_filled_qty(order: &mut OrderAny, filled_qty: Quantity) {
6380 match order {
6381 OrderAny::Limit(o) => o.filled_qty = filled_qty,
6382 OrderAny::LimitIfTouched(o) => o.filled_qty = filled_qty,
6383 OrderAny::Market(o) => o.filled_qty = filled_qty,
6384 OrderAny::MarketIfTouched(o) => o.filled_qty = filled_qty,
6385 OrderAny::MarketToLimit(o) => o.filled_qty = filled_qty,
6386 OrderAny::StopLimit(o) => o.filled_qty = filled_qty,
6387 OrderAny::StopMarket(o) => o.filled_qty = filled_qty,
6388 OrderAny::TrailingStopLimit(o) => o.filled_qty = filled_qty,
6389 OrderAny::TrailingStopMarket(o) => o.filled_qty = filled_qty,
6390 }
6391}
6392
6393#[derive(Debug, Clone, Copy)]
6394struct BarTickSizes {
6395 open: Quantity,
6396 high: Quantity,
6397 low: Quantity,
6398 close: Quantity,
6399}
6400
6401impl BarTickSizes {
6402 fn from_volume(volume: Quantity, size_increment: Quantity) -> Self {
6403 let precision_diff = FIXED_PRECISION.saturating_sub(volume.precision);
6404 let scale = QuantityRaw::pow(10, u32::from(precision_diff));
6405 let units = volume.raw / scale;
6406 let increment_units = (size_increment.raw / scale).max(1);
6407 let rounded_units = (units / increment_units) * increment_units;
6408 let increments = rounded_units / increment_units;
6409 let zero = Quantity::zero(volume.precision);
6410 let size =
6411 |increments| Quantity::from_raw(increments * increment_units * scale, volume.precision);
6412
6413 match increments {
6414 0 => Self {
6415 open: zero,
6416 high: zero,
6417 low: zero,
6418 close: zero,
6419 },
6420
6421 1 => Self {
6423 open: zero,
6424 high: zero,
6425 low: zero,
6426 close: size(1),
6427 },
6428 2 => Self {
6429 open: zero,
6430 high: size(1),
6431 low: size(1),
6432 close: zero,
6433 },
6434 3 => {
6435 let path_size = size(1);
6436
6437 Self {
6438 open: path_size,
6439 high: path_size,
6440 low: path_size,
6441 close: zero,
6442 }
6443 }
6444 _ => {
6445 let path_increments = increments / 4;
6446 let close_increments = increments - (path_increments * 3);
6447 let path_size = size(path_increments);
6448
6449 Self {
6450 open: path_size,
6451 high: path_size,
6452 low: path_size,
6453 close: size(close_increments),
6454 }
6455 }
6456 }
6457 }
6458}
6459
6460#[cfg(test)]
6461mod tests {
6462 use std::{
6463 cell::{Cell, RefCell},
6464 collections::{HashMap, HashSet},
6465 rc::Rc,
6466 };
6467
6468 use nautilus_common::{
6469 cache::Cache,
6470 clock::TestClock,
6471 messages::execution::{CancelAllOrders, ModifyOrder},
6472 };
6473 use nautilus_core::{UUID4, UnixNanos, correctness::CorrectnessError};
6474 use nautilus_model::{
6475 data::{
6476 DEPTH10_LEN, OrderBookDelta, OrderBookDepth10, QuoteTick, TradeTick,
6477 option_chain::OptionGreeks,
6478 order::{BookOrder, OrderId},
6479 },
6480 enums::{
6481 AccountType, AggressorSide, BookAction, BookType, LiquiditySide, OmsType, OrderSide,
6482 OrderStatus, OrderType, RecordFlag, TimeInForce, TrailingOffsetType, TriggerType,
6483 },
6484 events::OrderEventAny,
6485 identifiers::{AccountId, ClientOrderId, StrategyId, TradeId, TraderId, VenueOrderId},
6486 instruments::{
6487 Instrument, InstrumentAny,
6488 stubs::{crypto_option_btc_deribit, crypto_perpetual_ethusdt, futures_contract_es},
6489 },
6490 orderbook::OrderBook,
6491 orders::{Order, OrderAny, OrderTestBuilder, stubs::TestOrderEventStubs},
6492 types::{Money, Price, Quantity, fixed::FIXED_PRECISION, quantity::QuantityRaw},
6493 };
6494 use proptest::prelude::*;
6495 use rstest::rstest;
6496 use rust_decimal::Decimal;
6497
6498 use super::{BarTickSizes, OrderMatchingEngine, PostMatchOrderAction, post_match_order_action};
6499 use crate::{
6500 matching_engine::config::OrderMatchingEngineConfig,
6501 models::{
6502 fee::{FeeModel, FeeModelAny, FeeModelHandle},
6503 fill::{FillModel, FillModelHandle},
6504 },
6505 };
6506
6507 fn assert_valid_bar_tick_sizes(volume: Quantity, size_increment: Quantity) {
6508 let sizes = BarTickSizes::from_volume(volume, size_increment);
6509 let total_raw = sizes.open.raw + sizes.high.raw + sizes.low.raw + sizes.close.raw;
6510 assert!(total_raw <= volume.raw);
6511
6512 for quantity in [sizes.open, sizes.high, sizes.low, sizes.close] {
6513 assert_eq!(quantity.precision, volume.precision);
6514 assert!(
6515 OrderMatchingEngine::quantity_matches_precision(quantity, volume.precision),
6516 "bar tick quantity {quantity} not aligned to precision {}",
6517 volume.precision,
6518 );
6519 assert!(
6520 size_increment.raw == 0 || quantity.raw.is_multiple_of(size_increment.raw),
6521 "bar tick quantity {quantity} not aligned to increment {size_increment}",
6522 );
6523 }
6524
6525 if size_increment.raw > 0 {
6526 assert!(
6527 volume.raw - total_raw < size_increment.raw,
6528 "bar tick split left {} raw units from volume {volume} and increment {size_increment}",
6529 volume.raw - total_raw,
6530 );
6531 }
6532 }
6533
6534 #[rstest]
6535 fn test_post_match_order_action_does_not_clone_no_maintenance_order() {
6536 let order = post_match_limit_order();
6537 let clone_count = Cell::new(0);
6538
6539 let action = post_match_order_action(&order, true, UnixNanos::from(1_u64), |order| {
6540 clone_count.set(clone_count.get() + 1);
6541 order.clone()
6542 });
6543
6544 assert!(matches!(action, PostMatchOrderAction::NoMaintenance));
6545 assert_eq!(clone_count.get(), 0);
6546 }
6547
6548 #[rstest]
6549 fn test_post_match_order_action_does_not_clone_closed_order() {
6550 let order = post_match_closed_limit_order();
6551 let clone_count = Cell::new(0);
6552
6553 let action = post_match_order_action(&order, true, UnixNanos::from(1_u64), |order| {
6554 clone_count.set(clone_count.get() + 1);
6555 order.clone()
6556 });
6557
6558 assert!(matches!(action, PostMatchOrderAction::RemoveClosed));
6559 assert_eq!(clone_count.get(), 0);
6560 }
6561
6562 #[rstest]
6563 fn test_post_match_order_action_clones_expired_gtd_order_once() {
6564 let order = post_match_gtd_limit_order();
6565 let clone_count = Cell::new(0);
6566
6567 let action = post_match_order_action(&order, true, UnixNanos::from(10_u64), |order| {
6568 clone_count.set(clone_count.get() + 1);
6569 order.clone()
6570 });
6571
6572 let PostMatchOrderAction::Expire(cloned) = action else {
6573 panic!("Expected expired action, was {action:?}");
6574 };
6575 assert_eq!(cloned.client_order_id(), order.client_order_id());
6576 assert_eq!(clone_count.get(), 1);
6577 }
6578
6579 #[rstest]
6580 fn test_post_match_order_action_clones_trailing_order_once() {
6581 let order = post_match_trailing_stop_order();
6582 let clone_count = Cell::new(0);
6583
6584 let action = post_match_order_action(&order, true, UnixNanos::from(1_u64), |order| {
6585 clone_count.set(clone_count.get() + 1);
6586 order.clone()
6587 });
6588
6589 let PostMatchOrderAction::UpdateTrailing(cloned) = action else {
6590 panic!("Expected trailing update action, was {action:?}");
6591 };
6592 assert_eq!(cloned.client_order_id(), order.client_order_id());
6593 assert_eq!(clone_count.get(), 1);
6594 }
6595
6596 fn post_match_limit_order() -> OrderAny {
6597 OrderTestBuilder::new(OrderType::Limit)
6598 .instrument_id(crypto_perpetual_ethusdt().id())
6599 .side(OrderSide::Buy)
6600 .price(Price::from("1500.00"))
6601 .quantity(Quantity::from("1.000"))
6602 .client_order_id(ClientOrderId::from("POST-MATCH-LIMIT"))
6603 .submit(true)
6604 .build()
6605 }
6606
6607 fn post_match_closed_limit_order() -> OrderAny {
6608 let account_id = AccountId::from("SIM-001");
6609 let venue_order_id = VenueOrderId::from("V-001");
6610 let mut order = post_match_limit_order();
6611 order
6612 .apply(TestOrderEventStubs::accepted(
6613 &order,
6614 account_id,
6615 venue_order_id,
6616 ))
6617 .unwrap();
6618 order
6619 .apply(TestOrderEventStubs::canceled(
6620 &order,
6621 account_id,
6622 Some(venue_order_id),
6623 ))
6624 .unwrap();
6625 order
6626 }
6627
6628 fn post_match_gtd_limit_order() -> OrderAny {
6629 OrderTestBuilder::new(OrderType::Limit)
6630 .instrument_id(crypto_perpetual_ethusdt().id())
6631 .side(OrderSide::Buy)
6632 .price(Price::from("1500.00"))
6633 .quantity(Quantity::from("1.000"))
6634 .time_in_force(TimeInForce::Gtd)
6635 .expire_time(UnixNanos::from(10_u64))
6636 .client_order_id(ClientOrderId::from("POST-MATCH-GTD"))
6637 .submit(true)
6638 .build()
6639 }
6640
6641 fn post_match_trailing_stop_order() -> OrderAny {
6642 OrderTestBuilder::new(OrderType::TrailingStopMarket)
6643 .instrument_id(crypto_perpetual_ethusdt().id())
6644 .side(OrderSide::Buy)
6645 .quantity(Quantity::from("1.000"))
6646 .trigger_price(Price::from("1510.00"))
6647 .trigger_type(TriggerType::BidAsk)
6648 .trailing_offset(Decimal::new(5, 0))
6649 .trailing_offset_type(TrailingOffsetType::Price)
6650 .client_order_id(ClientOrderId::from("POST-MATCH-TRAIL"))
6651 .submit(true)
6652 .build()
6653 }
6654
6655 #[rstest]
6656 fn test_fill_order_calculates_commission_from_fill_liquidity_side() {
6657 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6658 let cache = Rc::new(RefCell::new(Cache::default()));
6659 let clock = Rc::new(RefCell::new(TestClock::new()));
6660 let mut engine = OrderMatchingEngine::new(
6661 instrument.clone(),
6662 1,
6663 FillModelHandle::default(),
6664 FeeModelAny::default().into(),
6665 BookType::L1_MBP,
6666 OmsType::Netting,
6667 AccountType::Margin,
6668 clock,
6669 cache,
6670 Default::default(),
6671 );
6672 let events = Rc::new(RefCell::new(Vec::new()));
6673 let events_handler = Rc::clone(&events);
6674 engine.set_event_handler(Rc::new(move |event| {
6675 events_handler.borrow_mut().push(event);
6676 }));
6677
6678 let mut order = OrderTestBuilder::new(OrderType::Market)
6679 .instrument_id(instrument.id())
6680 .side(OrderSide::Buy)
6681 .quantity(Quantity::from("1.000"))
6682 .submit(true)
6683 .build();
6684 order.set_liquidity_side(LiquiditySide::Maker);
6685 engine
6686 .account_ids
6687 .insert(order.trader_id(), AccountId::from("ACCOUNT-001"));
6688
6689 engine
6690 .fill_order(
6691 &order,
6692 Price::from("1500.00"),
6693 Quantity::from("1.000"),
6694 LiquiditySide::Taker,
6695 None,
6696 None,
6697 )
6698 .unwrap();
6699
6700 let events = events.borrow();
6701 assert_eq!(events.len(), 1);
6702 let fill = match &events[0] {
6703 OrderEventAny::Filled(fill) => fill,
6704 event => panic!("Expected OrderFilled, was {event:?}"),
6705 };
6706 let commission = fill.commission.expect("expected commission");
6707 let expected_commission =
6708 fill.last_qty.as_decimal() * fill.last_px.as_decimal() * instrument.taker_fee();
6709
6710 assert_eq!(fill.liquidity_side, LiquiditySide::Taker);
6711 assert_eq!(commission.currency, instrument.quote_currency());
6712 assert_eq!(commission.as_decimal(), expected_commission);
6713 }
6714
6715 #[rstest]
6716 fn test_custom_fee_model_handle_is_called_by_fill_order() {
6717 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6718 let cache = Rc::new(RefCell::new(Cache::default()));
6719 let clock = Rc::new(RefCell::new(TestClock::new()));
6720 let calls = Rc::new(Cell::new(0));
6721 let expected_commission = Money::from("1.23 USDT");
6722 let fee_model = FeeModelHandle::new(RecordingFeeModel {
6723 calls: Rc::clone(&calls),
6724 commission: expected_commission,
6725 });
6726 let cloned_fee_model = fee_model.clone();
6727 drop(fee_model);
6728 let mut engine = OrderMatchingEngine::new(
6729 instrument.clone(),
6730 1,
6731 FillModelHandle::default(),
6732 cloned_fee_model,
6733 BookType::L1_MBP,
6734 OmsType::Netting,
6735 AccountType::Margin,
6736 clock,
6737 cache,
6738 Default::default(),
6739 );
6740 let events = Rc::new(RefCell::new(Vec::new()));
6741 let events_handler = Rc::clone(&events);
6742 engine.set_event_handler(Rc::new(move |event| {
6743 events_handler.borrow_mut().push(event);
6744 }));
6745
6746 let order = OrderTestBuilder::new(OrderType::Market)
6747 .instrument_id(instrument.id())
6748 .side(OrderSide::Buy)
6749 .quantity(Quantity::from("1.000"))
6750 .submit(true)
6751 .build();
6752 engine
6753 .account_ids
6754 .insert(order.trader_id(), AccountId::from("ACCOUNT-001"));
6755
6756 engine
6757 .fill_order(
6758 &order,
6759 Price::from("1500.00"),
6760 Quantity::from("1.000"),
6761 LiquiditySide::Taker,
6762 None,
6763 None,
6764 )
6765 .unwrap();
6766
6767 let events = events.borrow();
6768 assert_eq!(events.len(), 1);
6769 let fill = match &events[0] {
6770 OrderEventAny::Filled(fill) => fill,
6771 event => panic!("Expected OrderFilled, was {event:?}"),
6772 };
6773
6774 assert_eq!(calls.get(), 1);
6775 assert_eq!(fill.commission, Some(expected_commission));
6776 }
6777
6778 #[rstest]
6779 fn test_fill_order_does_not_cache_filled_qty_when_fee_model_fails() {
6780 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6781 let cache = Rc::new(RefCell::new(Cache::default()));
6782 let clock = Rc::new(RefCell::new(TestClock::new()));
6783 let mut engine = OrderMatchingEngine::new(
6784 instrument.clone(),
6785 1,
6786 FillModelHandle::default(),
6787 FeeModelHandle::new(FailingFeeModel),
6788 BookType::L1_MBP,
6789 OmsType::Netting,
6790 AccountType::Margin,
6791 clock,
6792 cache,
6793 Default::default(),
6794 );
6795 let events = Rc::new(RefCell::new(Vec::new()));
6796 let events_handler = Rc::clone(&events);
6797 engine.set_event_handler(Rc::new(move |event| {
6798 events_handler.borrow_mut().push(event);
6799 }));
6800
6801 let order = OrderTestBuilder::new(OrderType::Market)
6802 .instrument_id(instrument.id())
6803 .side(OrderSide::Buy)
6804 .quantity(Quantity::from("1.000"))
6805 .submit(true)
6806 .build();
6807 engine
6808 .account_ids
6809 .insert(order.trader_id(), AccountId::from("ACCOUNT-001"));
6810
6811 let result = engine.fill_order(
6812 &order,
6813 Price::from("1500.00"),
6814 Quantity::from("1.000"),
6815 LiquiditySide::Taker,
6816 None,
6817 None,
6818 );
6819
6820 assert!(result.is_err());
6821 assert_eq!(engine.cached_filled_qty_len(), 0);
6822 assert!(events.borrow().is_empty());
6823 }
6824
6825 #[rstest]
6826 fn test_process_cancel_all_includes_submitted_orders_for_selected_account() {
6827 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6828 let instrument_id = instrument.id();
6829 let cache = Rc::new(RefCell::new(Cache::default()));
6830 let clock = Rc::new(RefCell::new(TestClock::new()));
6831 let mut engine = OrderMatchingEngine::new(
6832 instrument,
6833 1,
6834 FillModelHandle::default(),
6835 FeeModelAny::default().into(),
6836 BookType::L1_MBP,
6837 OmsType::Netting,
6838 AccountType::Margin,
6839 clock,
6840 Rc::clone(&cache),
6841 Default::default(),
6842 );
6843 let selected_account = AccountId::from("ACCOUNT-001");
6844 let other_account = AccountId::from("ACCOUNT-002");
6845 let selected_strategy = StrategyId::from("STRATEGY-001");
6846 let other_strategy = StrategyId::from("STRATEGY-002");
6847 let selected_order = OrderTestBuilder::new(OrderType::Limit)
6848 .strategy_id(selected_strategy)
6849 .instrument_id(instrument_id)
6850 .client_order_id(ClientOrderId::from("O-SUBMITTED-SELECTED"))
6851 .side(OrderSide::Buy)
6852 .price(Price::from("1400.00"))
6853 .quantity(Quantity::from("1.000"))
6854 .build();
6855 let other_order = OrderTestBuilder::new(OrderType::Limit)
6856 .strategy_id(other_strategy)
6857 .instrument_id(instrument_id)
6858 .client_order_id(ClientOrderId::from("O-SUBMITTED-OTHER"))
6859 .side(OrderSide::Buy)
6860 .price(Price::from("1300.00"))
6861 .quantity(Quantity::from("1.000"))
6862 .build();
6863 {
6864 let mut cache = cache.borrow_mut();
6865 cache
6866 .add_order(selected_order.clone(), None, None, false)
6867 .unwrap();
6868 cache
6869 .add_order(other_order.clone(), None, None, false)
6870 .unwrap();
6871 cache
6872 .update_order(&TestOrderEventStubs::submitted(
6873 &selected_order,
6874 selected_account,
6875 ))
6876 .unwrap();
6877 cache
6878 .update_order(&TestOrderEventStubs::submitted(&other_order, other_account))
6879 .unwrap();
6880 }
6881
6882 let events = Rc::new(RefCell::new(Vec::new()));
6883 let events_handler = Rc::clone(&events);
6884 let event_cache = Rc::clone(&cache);
6885 engine.set_event_handler(Rc::new(move |event| {
6886 event_cache.borrow_mut().update_order(&event).unwrap();
6887 events_handler.borrow_mut().push(event);
6888 }));
6889 let command = CancelAllOrders::new(
6890 TraderId::from("TRADER-001"),
6891 None,
6892 StrategyId::from("CALLER-001"),
6893 instrument_id,
6894 None,
6895 UUID4::new(),
6896 UnixNanos::default(),
6897 None,
6898 None,
6899 );
6900
6901 engine.process_cancel_all(&command, selected_account);
6902
6903 let events = events.borrow();
6904 assert_eq!(events.len(), 1);
6905 let OrderEventAny::Canceled(canceled) = &events[0] else {
6906 panic!("Expected OrderCanceled, was {:?}", events[0]);
6907 };
6908 assert_eq!(canceled.client_order_id, selected_order.client_order_id());
6909 assert_eq!(canceled.strategy_id, selected_strategy);
6910 assert_eq!(canceled.account_id, Some(selected_account));
6911 let cache = cache.borrow();
6912 assert_eq!(
6913 cache
6914 .order(&selected_order.client_order_id())
6915 .unwrap()
6916 .status(),
6917 OrderStatus::Canceled
6918 );
6919 assert_eq!(
6920 cache
6921 .order(&other_order.client_order_id())
6922 .unwrap()
6923 .status(),
6924 OrderStatus::Submitted
6925 );
6926 }
6927
6928 fn collision_engine() -> (OrderMatchingEngine, Rc<RefCell<Cache>>, VenueOrderId) {
6929 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6930 let cache = Rc::new(RefCell::new(Cache::default()));
6931 let venue_order_id = VenueOrderId::from(format!("{}-1-1", instrument.id().venue));
6932 cache
6933 .borrow_mut()
6934 .add_venue_order_id(&ClientOrderId::from("O-OWNER"), &venue_order_id, false)
6935 .unwrap();
6936 let engine = OrderMatchingEngine::new(
6937 instrument,
6938 1,
6939 FillModelHandle::default(),
6940 FeeModelAny::default().into(),
6941 BookType::L1_MBP,
6942 OmsType::Netting,
6943 AccountType::Margin,
6944 Rc::new(RefCell::new(TestClock::new())),
6945 Rc::clone(&cache),
6946 Default::default(),
6947 );
6948
6949 (engine, cache, venue_order_id)
6950 }
6951
6952 #[rstest]
6953 #[case(OrderType::Market)]
6954 #[case(OrderType::MarketToLimit)]
6955 fn test_market_collision_probes_and_fills_with_default_ack_config(
6956 #[case] order_type: OrderType,
6957 ) {
6958 let (mut engine, cache, venue_order_id) = collision_engine();
6959 assert!(!engine.config.use_market_order_acks);
6960 let quote = QuoteTick::new(
6961 engine.instrument.id(),
6962 Price::from("1499.00"),
6963 Price::from("1500.00"),
6964 Quantity::from("10.000"),
6965 Quantity::from("10.000"),
6966 UnixNanos::default(),
6967 UnixNanos::default(),
6968 );
6969 engine.process_quote_tick("e);
6970 let events = Rc::new(RefCell::new(Vec::new()));
6971 let events_handler = Rc::clone(&events);
6972 engine.set_event_handler(Rc::new(move |event| {
6973 events_handler.borrow_mut().push(event);
6974 }));
6975 let mut order = OrderTestBuilder::new(order_type)
6976 .instrument_id(engine.instrument.id())
6977 .client_order_id(ClientOrderId::from("O-CLAIMANT"))
6978 .side(OrderSide::Buy)
6979 .quantity(Quantity::from("1.000"))
6980 .submit(true)
6981 .build();
6982
6983 engine.process_order(&mut order, AccountId::from("ACCOUNT-001"));
6984
6985 assert!(
6986 !events
6987 .borrow()
6988 .iter()
6989 .any(|event| matches!(event, OrderEventAny::Rejected(_)))
6990 );
6991 assert!(
6992 events
6993 .borrow()
6994 .iter()
6995 .any(|event| matches!(event, OrderEventAny::Filled(_)))
6996 );
6997 assert!(cache.borrow().order_exists(&order.client_order_id()));
6998 assert_eq!(
6999 cache.borrow().client_order_id(&venue_order_id),
7000 Some(&ClientOrderId::from("O-OWNER"))
7001 );
7002 assert_eq!(
7003 cache.borrow().venue_order_id(&order.client_order_id()),
7004 Some(&VenueOrderId::from(format!("{}-1-2", engine.venue)))
7005 );
7006 }
7007
7008 struct RecordingFeeModel {
7009 calls: Rc<Cell<u32>>,
7010 commission: Money,
7011 }
7012
7013 impl FeeModel for RecordingFeeModel {
7014 fn get_commission(
7015 &self,
7016 _order: &OrderAny,
7017 _fill_quantity: Quantity,
7018 _fill_px: Price,
7019 _instrument: &InstrumentAny,
7020 ) -> anyhow::Result<Money> {
7021 self.calls.set(self.calls.get() + 1);
7022 Ok(self.commission)
7023 }
7024 }
7025
7026 struct FailingFeeModel;
7027
7028 impl FeeModel for FailingFeeModel {
7029 fn get_commission(
7030 &self,
7031 _order: &OrderAny,
7032 _fill_quantity: Quantity,
7033 _fill_px: Price,
7034 _instrument: &InstrumentAny,
7035 ) -> anyhow::Result<Money> {
7036 Err(anyhow::anyhow!("fee model failed"))
7037 }
7038 }
7039
7040 #[rstest]
7041 fn test_custom_fill_model_handle_is_called_by_market_fill() {
7042 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7043 let cache = Rc::new(RefCell::new(Cache::default()));
7044 let clock = Rc::new(RefCell::new(TestClock::new()));
7045 let calls = Rc::new(Cell::new(0));
7046 let fill_model = FillModelHandle::new(RecordingFillModel {
7047 calls: Rc::clone(&calls),
7048 });
7049 let mut engine = OrderMatchingEngine::new(
7050 instrument.clone(),
7051 1,
7052 fill_model,
7053 FeeModelAny::default().into(),
7054 BookType::L1_MBP,
7055 OmsType::Netting,
7056 AccountType::Margin,
7057 clock,
7058 cache,
7059 Default::default(),
7060 );
7061 let quote = QuoteTick::new(
7062 instrument.id(),
7063 Price::from("1500.00"),
7064 Price::from("1501.00"),
7065 Quantity::from("10.000"),
7066 Quantity::from("10.000"),
7067 UnixNanos::default(),
7068 UnixNanos::default(),
7069 );
7070 engine.process_quote_tick("e);
7071
7072 let mut order = OrderTestBuilder::new(OrderType::Market)
7073 .instrument_id(instrument.id())
7074 .side(OrderSide::Buy)
7075 .quantity(Quantity::from("1.000"))
7076 .submit(true)
7077 .build();
7078 engine.process_order(&mut order, AccountId::from("ACCOUNT-001"));
7079
7080 assert_eq!(calls.get(), 1);
7081 }
7082
7083 #[rstest]
7084 fn test_l1_depth10_skips_padding_for_last_quote_tracking() {
7085 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7086 let cache = Rc::new(RefCell::new(Cache::default()));
7087 let clock = Rc::new(RefCell::new(TestClock::new()));
7088 let mut engine = OrderMatchingEngine::new(
7089 instrument.clone(),
7090 1,
7091 FillModelHandle::default(),
7092 FeeModelAny::default().into(),
7093 BookType::L1_MBP,
7094 OmsType::Netting,
7095 AccountType::Margin,
7096 clock,
7097 cache,
7098 Default::default(),
7099 );
7100 let mut bids = [BookOrder::default(); DEPTH10_LEN];
7101 let mut asks = [BookOrder::default(); DEPTH10_LEN];
7102 bids[1] = BookOrder::new(
7103 OrderSide::Buy,
7104 Price::from("1499.00"),
7105 Quantity::from("1.000"),
7106 1,
7107 );
7108 asks[0] = BookOrder::new(
7109 OrderSide::Sell,
7110 Price::from("1500.00"),
7111 Quantity::from("1.000"),
7112 2,
7113 );
7114
7115 let depth = OrderBookDepth10::new(
7116 instrument.id(),
7117 bids,
7118 asks,
7119 [0; DEPTH10_LEN],
7120 [0; DEPTH10_LEN],
7121 0,
7122 0,
7123 UnixNanos::from(1_u64),
7124 UnixNanos::from(1_u64),
7125 );
7126 engine.process_order_book_depth10(&depth).unwrap();
7127
7128 assert_eq!(engine.last_quote_bid, Some(Price::from("1499.00")));
7129 assert_eq!(engine.last_quote_ask, Some(Price::from("1500.00")));
7130
7131 let depth_without_bid = OrderBookDepth10::new(
7132 instrument.id(),
7133 [BookOrder::default(); DEPTH10_LEN],
7134 asks,
7135 [0; DEPTH10_LEN],
7136 [0; DEPTH10_LEN],
7137 0,
7138 1,
7139 UnixNanos::from(2_u64),
7140 UnixNanos::from(2_u64),
7141 );
7142 engine
7143 .process_order_book_depth10(&depth_without_bid)
7144 .unwrap();
7145
7146 assert_eq!(engine.last_quote_bid, None);
7147 assert_eq!(engine.last_quote_ask, Some(Price::from("1500.00")));
7148 }
7149
7150 struct RecordingFillModel {
7151 calls: Rc<Cell<u32>>,
7152 }
7153
7154 impl FillModel for RecordingFillModel {
7155 fn is_limit_filled(&mut self) -> anyhow::Result<bool> {
7156 Ok(true)
7157 }
7158
7159 fn is_slipped(&mut self) -> anyhow::Result<bool> {
7160 Ok(false)
7161 }
7162
7163 fn get_orderbook_for_fill_simulation(
7164 &mut self,
7165 _instrument: &InstrumentAny,
7166 _order: &OrderAny,
7167 _best_bid: Price,
7168 _best_ask: Price,
7169 ) -> anyhow::Result<Option<OrderBook>> {
7170 self.calls.set(self.calls.get() + 1);
7171 Ok(None)
7172 }
7173 }
7174
7175 #[rstest]
7176 fn test_fee_underlying_price_uses_valid_cached_greeks_price() {
7177 let instrument = InstrumentAny::CryptoOption(crypto_option_btc_deribit(
7178 3,
7179 1,
7180 Price::from("0.001"),
7181 Quantity::from("0.1"),
7182 ));
7183 let cache = Rc::new(RefCell::new(Cache::default()));
7184 cache.borrow_mut().add_option_greeks(OptionGreeks {
7185 instrument_id: instrument.id(),
7186 underlying_price: Some(50_000.0),
7187 ..Default::default()
7188 });
7189 let clock = Rc::new(RefCell::new(TestClock::new()));
7190 let engine = OrderMatchingEngine::new(
7191 instrument,
7192 1,
7193 FillModelHandle::default(),
7194 FeeModelAny::default().into(),
7195 BookType::L1_MBP,
7196 OmsType::Netting,
7197 AccountType::Margin,
7198 clock,
7199 cache,
7200 Default::default(),
7201 );
7202
7203 let price = engine
7204 .fee_underlying_price()
7205 .unwrap()
7206 .expect("expected underlying price");
7207
7208 assert_eq!(price.precision, FIXED_PRECISION);
7209 assert_eq!(price.as_decimal(), Decimal::from(50_000));
7210 }
7211
7212 #[rstest]
7213 fn test_fee_underlying_price_rejects_invalid_cached_greeks_price() {
7214 let instrument = InstrumentAny::CryptoOption(crypto_option_btc_deribit(
7215 3,
7216 1,
7217 Price::from("0.001"),
7218 Quantity::from("0.1"),
7219 ));
7220 let cache = Rc::new(RefCell::new(Cache::default()));
7221 cache.borrow_mut().add_option_greeks(OptionGreeks {
7222 instrument_id: instrument.id(),
7223 underlying_price: Some(f64::NAN),
7224 ..Default::default()
7225 });
7226 let clock = Rc::new(RefCell::new(TestClock::new()));
7227 let engine = OrderMatchingEngine::new(
7228 instrument,
7229 1,
7230 FillModelHandle::default(),
7231 FeeModelAny::default().into(),
7232 BookType::L1_MBP,
7233 OmsType::Netting,
7234 AccountType::Margin,
7235 clock,
7236 cache,
7237 Default::default(),
7238 );
7239
7240 let error = engine.fee_underlying_price().unwrap_err();
7241
7242 assert_eq!(
7243 error,
7244 CorrectnessError::InvalidValue {
7245 param: "value".to_string(),
7246 value: "NaN".to_string(),
7247 type_name: "f64",
7248 }
7249 );
7250 }
7251
7252 #[rstest]
7253 fn test_bar_tick_sizes_divisible() {
7254 let volume = Quantity::from("100.000");
7256 let increment = Quantity::from("0.001");
7257 let sizes = BarTickSizes::from_volume(volume, increment);
7258 assert_eq!(sizes.open, Quantity::from("25.000"));
7259 assert_eq!(sizes.high, Quantity::from("25.000"));
7260 assert_eq!(sizes.low, Quantity::from("25.000"));
7261 assert_eq!(sizes.close, Quantity::from("25.000"));
7262 assert_valid_bar_tick_sizes(volume, increment);
7263 }
7264
7265 #[rstest]
7266 fn test_bar_tick_sizes_indivisible_with_remainder() {
7267 let volume = Quantity::from("0.05");
7269 let increment = Quantity::from("0.01");
7270 let sizes = BarTickSizes::from_volume(volume, increment);
7271 assert_eq!(sizes.open, Quantity::from("0.01"));
7272 assert_eq!(sizes.high, Quantity::from("0.01"));
7273 assert_eq!(sizes.low, Quantity::from("0.01"));
7274 assert_eq!(sizes.close, Quantity::from("0.02"));
7275 assert_valid_bar_tick_sizes(volume, increment);
7276 assert_eq!(
7277 sizes.open.raw + sizes.high.raw + sizes.low.raw + sizes.close.raw,
7278 volume.raw
7279 );
7280 }
7281
7282 #[rstest]
7283 #[case("1", "0", "0", "0", "1")]
7284 #[case("2", "0", "1", "1", "0")]
7285 #[case("3", "1", "1", "1", "0")]
7286 fn test_bar_tick_sizes_units_less_than_four_preserves_volume(
7287 #[case] volume: &str,
7288 #[case] open_size: &str,
7289 #[case] high_size: &str,
7290 #[case] low_size: &str,
7291 #[case] close_size: &str,
7292 ) {
7293 let volume = Quantity::from(volume);
7294 let increment = Quantity::from("1");
7295 let sizes = BarTickSizes::from_volume(volume, increment);
7296
7297 assert_eq!(sizes.open, Quantity::from(open_size));
7298 assert_eq!(sizes.high, Quantity::from(high_size));
7299 assert_eq!(sizes.low, Quantity::from(low_size));
7300 assert_eq!(sizes.close, Quantity::from(close_size));
7301 assert_valid_bar_tick_sizes(volume, increment);
7302 assert_eq!(
7303 sizes.open.raw + sizes.high.raw + sizes.low.raw + sizes.close.raw,
7304 volume.raw
7305 );
7306 }
7307
7308 #[rstest]
7309 fn test_bar_tick_sizes_zero_volume_remains_zero() {
7310 let volume = Quantity::zero(3);
7311 let increment = Quantity::from("0.001");
7312 let sizes = BarTickSizes::from_volume(volume, increment);
7313 assert_eq!(sizes.open, Quantity::zero(3));
7314 assert_eq!(sizes.high, Quantity::zero(3));
7315 assert_eq!(sizes.low, Quantity::zero(3));
7316 assert_eq!(sizes.close, Quantity::zero(3));
7317 assert_valid_bar_tick_sizes(volume, increment);
7318 }
7319
7320 #[rstest]
7321 fn test_bar_tick_sizes_rounds_down_to_size_increment() {
7322 let volume = Quantity::from("1.07");
7323 let increment = Quantity::from("0.10");
7324 let sizes = BarTickSizes::from_volume(volume, increment);
7325 assert_eq!(sizes.open, Quantity::from("0.20"));
7326 assert_eq!(sizes.high, Quantity::from("0.20"));
7327 assert_eq!(sizes.low, Quantity::from("0.20"));
7328 assert_eq!(sizes.close, Quantity::from("0.40"));
7329 assert_valid_bar_tick_sizes(volume, increment);
7330 }
7331
7332 #[rstest]
7333 fn test_bar_tick_sizes_at_fixed_precision() {
7334 let units: QuantityRaw = 17;
7337 let volume = Quantity::from_raw(units, FIXED_PRECISION);
7338 let increment = Quantity::from_raw(1, FIXED_PRECISION);
7339 let sizes = BarTickSizes::from_volume(volume, increment);
7340 assert_eq!(sizes.open.raw, 4);
7341 assert_eq!(sizes.high.raw, 4);
7342 assert_eq!(sizes.low.raw, 4);
7343 assert_eq!(sizes.close.raw, 5);
7344 assert_valid_bar_tick_sizes(volume, increment);
7345 }
7346
7347 fn get_l3_queue_engine(instrument: InstrumentAny) -> (OrderMatchingEngine, Rc<RefCell<Cache>>) {
7348 let clock = Rc::new(RefCell::new(TestClock::new()));
7349 let cache = Rc::new(RefCell::new(Cache::default()));
7350 let config = OrderMatchingEngineConfig {
7351 trade_execution: true,
7352 queue_position: true,
7353 ..Default::default()
7354 };
7355
7356 let mut engine = OrderMatchingEngine::new(
7357 instrument,
7358 1,
7359 FillModelHandle::default(),
7360 FeeModelAny::default().into(),
7361 BookType::L3_MBO,
7362 OmsType::Netting,
7363 AccountType::Margin,
7364 clock,
7365 Rc::clone(&cache),
7366 config,
7367 );
7368
7369 let handler_cache = Rc::clone(&cache);
7370 engine.set_event_handler(Rc::new(move |event: OrderEventAny| {
7371 if let Ok(mut cache) = handler_cache.try_borrow_mut() {
7372 let _ = cache.update_order(&event);
7373 }
7374 }));
7375
7376 (engine, cache)
7377 }
7378
7379 fn assert_l3_queue_synced(engine: &OrderMatchingEngine) {
7380 for (client_order_id, orders_ahead) in &engine.queue_ahead_orders {
7381 let set_sum: QuantityRaw = orders_ahead.values().sum();
7382 let counter = engine
7383 .queue_ahead_total
7384 .get(client_order_id)
7385 .map_or(0, |&(_, ahead_raw)| ahead_raw);
7386 assert_eq!(
7387 set_sum, counter,
7388 "tracked orders out of sync with quantity-ahead counter for {client_order_id}",
7389 );
7390 }
7391
7392 for (client_order_id, price_raw) in &engine.queue_pending {
7393 assert!(
7394 engine
7395 .queue_ids_by_price
7396 .get(price_raw)
7397 .is_some_and(|ids| ids.contains(client_order_id)),
7398 "pending order {client_order_id} missing from price index",
7399 );
7400 }
7401
7402 for (client_order_id, (price_raw, _)) in &engine.queue_ahead_total {
7403 assert!(
7404 engine
7405 .queue_ids_by_price
7406 .get(price_raw)
7407 .is_some_and(|ids| ids.contains(client_order_id)),
7408 "tracked order {client_order_id} missing from price index",
7409 );
7410 }
7411
7412 for (price_raw, client_order_ids) in &engine.queue_ids_by_price {
7413 for client_order_id in client_order_ids {
7414 let pending_at_price = engine.queue_pending.get(client_order_id) == Some(price_raw);
7415 let tracked_at_price = engine
7416 .queue_ahead_total
7417 .get(client_order_id)
7418 .is_some_and(|(tracked_price_raw, _)| tracked_price_raw == price_raw);
7419 assert!(
7420 pending_at_price || tracked_at_price,
7421 "price index contains stale order {client_order_id}",
7422 );
7423 }
7424 }
7425 }
7426
7427 #[rstest]
7428 fn test_reset_clears_queue_positions() {
7429 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7430 let (mut engine, _cache) = get_l3_queue_engine(instrument);
7431 let price = Price::from("100.00");
7432 let client_order_id = ClientOrderId::from("O-RESET-QUEUE");
7433
7434 rest_l3_queue_order(&mut engine, price, 1, client_order_id);
7435
7436 assert!(engine.queue_ahead_total.contains_key(&client_order_id));
7437 assert!(engine.queue_ahead_orders.contains_key(&client_order_id));
7438 assert!(
7439 engine
7440 .queue_ids_by_price
7441 .get(&price.raw)
7442 .is_some_and(|ids| ids.contains(&client_order_id)),
7443 );
7444
7445 engine.reset();
7446
7447 assert!(engine.queue_pending.is_empty());
7448 assert!(engine.queue_ahead_total.is_empty());
7449 assert!(engine.queue_ahead_orders.is_empty());
7450 assert!(engine.queue_excess.is_empty());
7451 assert!(engine.queue_ids_by_price.is_empty());
7452 }
7453
7454 #[rstest]
7455 fn test_cancel_removes_queue_position() {
7456 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7457 let (mut engine, _cache) = get_l3_queue_engine(instrument);
7458 let price = Price::from("100.00");
7459 let order =
7460 rest_l3_queue_order(&mut engine, price, 1, ClientOrderId::from("O-CANCEL-QUEUE"));
7461 let client_order_id = order.client_order_id();
7462
7463 assert!(engine.queue_ahead_total.contains_key(&client_order_id));
7464 assert!(engine.queue_ahead_orders.contains_key(&client_order_id));
7465 assert!(
7466 engine
7467 .queue_ids_by_price
7468 .get(&price.raw)
7469 .is_some_and(|ids| ids.contains(&client_order_id)),
7470 );
7471
7472 engine.cancel_order(&order, None);
7473
7474 assert!(!engine.queue_pending.contains_key(&client_order_id));
7475 assert!(!engine.queue_ahead_total.contains_key(&client_order_id));
7476 assert!(!engine.queue_ahead_orders.contains_key(&client_order_id));
7477 assert!(!engine.queue_excess.contains_key(&client_order_id));
7478 assert!(!engine.queue_ids_by_price.contains_key(&price.raw));
7479 }
7480
7481 #[rstest]
7482 fn test_modify_reindexes_queue_position() {
7483 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7484 let (mut engine, _cache) = get_l3_queue_engine(instrument);
7485 let old_price = Price::from("100.00");
7486 let new_price = Price::from("101.00");
7487 let client_order_id = ClientOrderId::from("O-MODIFY-QUEUE");
7488 let order = rest_l3_queue_order(&mut engine, old_price, 1, client_order_id);
7489 let new_level = OrderBookDelta::new(
7490 engine.instrument.id(),
7491 BookAction::Add,
7492 BookOrder::new(OrderSide::Sell, new_price, Quantity::from("10.000"), 2),
7493 0,
7494 2,
7495 UnixNanos::from(2),
7496 UnixNanos::from(2),
7497 );
7498 engine.process_order_book_delta(&new_level).unwrap();
7499
7500 let command = ModifyOrder::new(
7501 order.trader_id(),
7502 None,
7503 order.strategy_id(),
7504 order.instrument_id(),
7505 client_order_id,
7506 order.venue_order_id(),
7507 None,
7508 Some(new_price),
7509 None,
7510 UUID4::new(),
7511 UnixNanos::from(3),
7512 None,
7513 None,
7514 );
7515 engine.process_modify(&command, AccountId::from("SIM-001"));
7516
7517 assert!(!engine.queue_ids_by_price.contains_key(&old_price.raw));
7518 assert_eq!(
7519 engine
7520 .queue_ids_by_price
7521 .get(&new_price.raw)
7522 .map(|ids| ids.iter().copied().collect::<Vec<_>>()),
7523 Some(vec![client_order_id]),
7524 );
7525 assert_eq!(
7526 engine.queue_ahead_total.get(&client_order_id),
7527 Some(&(new_price.raw, Quantity::from("10.000").raw)),
7528 );
7529 assert_eq!(
7530 engine
7531 .queue_ahead_orders
7532 .get(&client_order_id)
7533 .map(|orders| orders.keys().copied().collect::<Vec<_>>()),
7534 Some(vec![2]),
7535 );
7536 }
7537
7538 #[rstest]
7539 fn test_queue_price_index_filters_other_prices() {
7540 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7541 let (mut engine, _cache) = get_l3_queue_engine(instrument);
7542 let target_price = Price::from("100.00");
7543 let other_price = Price::from("101.00");
7544 let target_id = ClientOrderId::from("O-QUEUE-TARGET");
7545 let other_id = ClientOrderId::from("O-QUEUE-OTHER");
7546
7547 rest_l3_queue_order(&mut engine, target_price, 1, target_id);
7548 rest_l3_queue_order(&mut engine, other_price, 2, other_id);
7549
7550 let indexed_ids = engine.take_queue_ids_at_price(target_price.raw);
7551
7552 assert_eq!(indexed_ids, vec![target_id]);
7553 assert!(
7554 engine
7555 .queue_ids_by_price
7556 .get(&other_price.raw)
7557 .is_some_and(|ids| ids.contains(&other_id)),
7558 );
7559 }
7560
7561 fn rest_l3_queue_order(
7562 engine: &mut OrderMatchingEngine,
7563 price: Price,
7564 sequence: u64,
7565 client_order_id: ClientOrderId,
7566 ) -> OrderAny {
7567 let instrument_id = engine.instrument.id();
7568 let delta = OrderBookDelta::new(
7569 instrument_id,
7570 BookAction::Add,
7571 BookOrder::new(OrderSide::Sell, price, Quantity::from("10.000"), sequence),
7572 0,
7573 sequence,
7574 UnixNanos::from(sequence),
7575 UnixNanos::from(sequence),
7576 );
7577 engine.process_order_book_delta(&delta).unwrap();
7578
7579 let mut order = OrderTestBuilder::new(OrderType::Limit)
7580 .instrument_id(instrument_id)
7581 .side(OrderSide::Sell)
7582 .price(price)
7583 .quantity(Quantity::from("5.000"))
7584 .client_order_id(client_order_id)
7585 .submit(true)
7586 .build();
7587 engine.process_order(&mut order, AccountId::from("SIM-001"));
7588
7589 order
7590 }
7591
7592 #[derive(Debug, Clone, Copy)]
7593 enum QueueEvent {
7594 Add { id: OrderId, size: u64 },
7595 Update { id: OrderId, size: u64 },
7596 MoveAway { id: OrderId },
7597 Delete { id: OrderId },
7598 Trade { size: u64, aggressor: u8 },
7599 AggregateCap { size: u64 },
7600 AggregateDelete,
7601 RestOrder,
7602 }
7603
7604 fn granular_queue_event() -> impl Strategy<Value = QueueEvent> {
7605 prop_oneof![
7606 3 => (1u64..=6, 1u64..=9).prop_map(|(id, size)| QueueEvent::Add { id, size }),
7607 3 => (1u64..=6, 1u64..=9).prop_map(|(id, size)| QueueEvent::Update { id, size }),
7608 1 => (1u64..=6).prop_map(|id| QueueEvent::MoveAway { id }),
7609 2 => (1u64..=6).prop_map(|id| QueueEvent::Delete { id }),
7610 2 => Just(QueueEvent::RestOrder),
7611 ]
7612 }
7613
7614 fn any_queue_event() -> impl Strategy<Value = QueueEvent> {
7615 prop_oneof![
7616 5 => granular_queue_event(),
7617 3 => (1u64..=9, 0u8..3).prop_map(|(size, aggressor)| QueueEvent::Trade {
7618 size,
7619 aggressor,
7620 }),
7621 1 => (1u64..=9).prop_map(|size| QueueEvent::AggregateCap { size }),
7622 1 => Just(QueueEvent::AggregateDelete),
7623 ]
7624 }
7625
7626 struct L3QueueSim {
7629 engine: OrderMatchingEngine,
7630 account_id: AccountId,
7631 live_main: HashMap<OrderId, u64>,
7632 live_away: HashSet<OrderId>,
7633 rest_snapshots: HashMap<ClientOrderId, HashSet<OrderId>>,
7634 rested: usize,
7635 sequence: u64,
7636 }
7637
7638 impl L3QueueSim {
7639 const MAIN_PRICE: &'static str = "100.00";
7640 const AWAY_PRICE: &'static str = "101.00";
7641
7642 fn new() -> Self {
7643 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
7644 let (engine, _cache) = get_l3_queue_engine(instrument);
7645
7646 Self {
7647 engine,
7648 account_id: AccountId::from("SIM-001"),
7649 live_main: HashMap::new(),
7650 live_away: HashSet::new(),
7651 rest_snapshots: HashMap::new(),
7652 rested: 0,
7653 sequence: 0,
7654 }
7655 }
7656
7657 fn quantity(size: u64) -> Quantity {
7658 Quantity::from(format!("{size}.000").as_str())
7659 }
7660
7661 fn process_delta(
7662 &mut self,
7663 action: BookAction,
7664 price: &str,
7665 size: u64,
7666 order_id: OrderId,
7667 flags: u8,
7668 ) {
7669 self.sequence += 1;
7670 let delta = OrderBookDelta::new(
7671 self.engine.instrument.id(),
7672 action,
7673 BookOrder::new(
7674 OrderSide::Sell,
7675 Price::from(price),
7676 Self::quantity(size),
7677 order_id,
7678 ),
7679 flags,
7680 self.sequence,
7681 UnixNanos::from(self.sequence),
7682 UnixNanos::from(self.sequence),
7683 );
7684 self.engine.process_order_book_delta(&delta).unwrap();
7685 }
7686
7687 fn apply(&mut self, event: QueueEvent) {
7688 match event {
7689 QueueEvent::Add { id, size } => {
7690 if self.live_main.contains_key(&id) || self.live_away.contains(&id) {
7691 return;
7692 }
7693 self.process_delta(BookAction::Add, Self::MAIN_PRICE, size, id, 0);
7694 self.live_main.insert(id, size);
7695 }
7696 QueueEvent::Update { id, size } => {
7697 if !self.live_main.contains_key(&id) {
7698 return;
7699 }
7700 self.process_delta(BookAction::Update, Self::MAIN_PRICE, size, id, 0);
7701 self.live_main.insert(id, size);
7702 }
7703 QueueEvent::MoveAway { id } => {
7704 let Some(size) = self.live_main.remove(&id) else {
7705 return;
7706 };
7707 self.process_delta(BookAction::Update, Self::AWAY_PRICE, size, id, 0);
7708 self.live_away.insert(id);
7709 }
7710 QueueEvent::Delete { id } => {
7711 if let Some(size) = self.live_main.remove(&id) {
7712 self.process_delta(BookAction::Delete, Self::MAIN_PRICE, size, id, 0);
7713 } else if self.live_away.remove(&id) {
7714 self.process_delta(BookAction::Delete, Self::AWAY_PRICE, 1, id, 0);
7715 } else {
7716 self.process_delta(BookAction::Delete, Self::MAIN_PRICE, 1, id, 0);
7718 }
7719
7720 for snapshot_ids in self.rest_snapshots.values_mut() {
7723 snapshot_ids.remove(&id);
7724 }
7725 }
7726 QueueEvent::Trade { size, aggressor } => {
7727 self.sequence += 1;
7728 let aggressor_side = match aggressor {
7729 0 => AggressorSide::Buy,
7730 1 => AggressorSide::Sell,
7731 _ => AggressorSide::NoAggressor,
7732 };
7733 let trade = TradeTick::new(
7734 self.engine.instrument.id(),
7735 Price::from(Self::MAIN_PRICE),
7736 Self::quantity(size),
7737 aggressor_side,
7738 TradeId::new(format!("T-{}", self.sequence).as_str()),
7739 UnixNanos::from(self.sequence),
7740 UnixNanos::from(self.sequence),
7741 );
7742 self.engine.process_trade_tick(&trade);
7743 }
7744 QueueEvent::AggregateCap { size } => {
7745 self.process_delta(
7746 BookAction::Update,
7747 Self::MAIN_PRICE,
7748 size,
7749 0,
7750 RecordFlag::F_MBP as u8,
7751 );
7752 }
7753 QueueEvent::AggregateDelete => {
7754 self.process_delta(
7755 BookAction::Delete,
7756 Self::MAIN_PRICE,
7757 1,
7758 0,
7759 RecordFlag::F_MBP as u8,
7760 );
7761 }
7762 QueueEvent::RestOrder => {
7763 if self.rested >= 3 {
7764 return;
7765 }
7766 self.rested += 1;
7767 let mut order = OrderTestBuilder::new(OrderType::Limit)
7768 .instrument_id(self.engine.instrument.id())
7769 .side(OrderSide::Sell)
7770 .price(Price::from(Self::MAIN_PRICE))
7771 .quantity(Self::quantity(5))
7772 .client_order_id(ClientOrderId::from(
7773 format!("O-PROP-{}", self.rested).as_str(),
7774 ))
7775 .submit(true)
7776 .build();
7777 self.engine.process_order(&mut order, self.account_id);
7778
7779 assert!(
7780 self.engine
7781 .queue_ahead_orders
7782 .contains_key(&order.client_order_id()),
7783 "L3 snapshot must track the resting order",
7784 );
7785
7786 self.rest_snapshots.insert(
7787 order.client_order_id(),
7788 self.live_main.keys().copied().collect(),
7789 );
7790 }
7791 }
7792 }
7793
7794 fn assert_tracked_orders_match_book(&self) {
7797 let level: HashMap<OrderId, QuantityRaw> = self
7798 .engine
7799 .book
7800 .get_orders_at_level(Price::from(Self::MAIN_PRICE), OrderSide::Buy)
7801 .iter()
7802 .map(|order| (order.order_id, order.size.raw))
7803 .collect();
7804
7805 for (client_order_id, orders_ahead) in &self.engine.queue_ahead_orders {
7806 for (order_id, size_raw) in orders_ahead {
7807 let book_size = level.get(order_id).copied().unwrap_or_else(|| {
7808 panic!("tracked order {order_id} for {client_order_id} not in book level")
7809 });
7810 assert_eq!(
7811 book_size, *size_raw,
7812 "tracked size diverged from book for order {order_id}",
7813 );
7814 }
7815
7816 let tracked: HashSet<OrderId> = orders_ahead.keys().copied().collect();
7817 let expected: HashSet<OrderId> = self.rest_snapshots[client_order_id]
7818 .iter()
7819 .filter(|id| self.live_main.contains_key(id))
7820 .copied()
7821 .collect();
7822 assert_eq!(
7823 tracked, expected,
7824 "tracked set incomplete or stale for {client_order_id}",
7825 );
7826 }
7827 }
7828 }
7829
7830 #[rstest]
7831 fn prop_test_l3_queue_tracking_stays_synced_with_counter() {
7832 proptest!(|(events in prop::collection::vec(any_queue_event(), 1..=80))| {
7833 let mut sim = L3QueueSim::new();
7834 for event in events {
7835 sim.apply(event);
7836 assert_l3_queue_synced(&sim.engine);
7837 }
7838 });
7839 }
7840
7841 #[rstest]
7842 fn prop_test_l3_queue_tracking_mirrors_book_without_trades() {
7843 proptest!(|(events in prop::collection::vec(granular_queue_event(), 1..=80))| {
7844 let mut sim = L3QueueSim::new();
7845 for event in events {
7846 sim.apply(event);
7847 assert_l3_queue_synced(&sim.engine);
7848 sim.assert_tracked_orders_match_book();
7849 }
7850 });
7851 }
7852
7853 #[rstest]
7858 fn test_l3_queue_position_replay_databento_mbo_stays_synced() {
7859 let json = include_str!("../../../../test_data/databento/esh4-glbx-mdp3-20231225.mbo.json");
7860 let records: Vec<serde_json::Value> = serde_json::from_str(json).unwrap();
7861 assert!(records.len() > 1000);
7862
7863 let instrument = InstrumentAny::FuturesContract(futures_contract_es(None, None));
7864 let instrument_id = instrument.id();
7865 let (mut engine, cache) = get_l3_queue_engine(instrument);
7866 let account_id = AccountId::from("SIM-001");
7867
7868 let mut rested = 0usize;
7869 let mut trades = 0usize;
7870
7871 for (index, record) in records.iter().enumerate() {
7872 match record.get("type").and_then(serde_json::Value::as_str) {
7873 Some("OrderBookDelta") => {
7874 let mut delta: OrderBookDelta = serde_json::from_value(record.clone()).unwrap();
7875 delta.instrument_id = instrument_id;
7876 engine.process_order_book_delta(&delta).unwrap();
7877 }
7878 Some("TradeTick") => {
7879 let mut trade: TradeTick = serde_json::from_value(record.clone()).unwrap();
7880 trade.instrument_id = instrument_id;
7881 engine.process_trade_tick(&trade);
7882 trades += 1;
7883 }
7884 other => panic!("unexpected record type {other:?}"),
7885 }
7886
7887 if index % 150 == 100 {
7888 let (side, price) = if rested.is_multiple_of(2) {
7889 (OrderSide::Sell, engine.book.best_ask_price())
7890 } else {
7891 (OrderSide::Buy, engine.book.best_bid_price())
7892 };
7893
7894 if let Some(price) = price {
7895 rested += 1;
7896 let mut order = OrderTestBuilder::new(OrderType::Limit)
7897 .instrument_id(instrument_id)
7898 .side(side)
7899 .price(price)
7900 .quantity(Quantity::from("1"))
7901 .client_order_id(ClientOrderId::from(format!("O-MBO-{rested}").as_str()))
7902 .submit(true)
7903 .build();
7904 engine.process_order(&mut order, account_id);
7905
7906 let is_open = cache
7909 .borrow()
7910 .order(&order.client_order_id())
7911 .is_some_and(|order| order.is_open());
7912 if is_open {
7913 assert!(
7914 engine
7915 .queue_ahead_orders
7916 .contains_key(&order.client_order_id()),
7917 "L3 snapshot must track the resting order",
7918 );
7919 }
7920 }
7921 }
7922
7923 assert_l3_queue_synced(&engine);
7924 }
7925
7926 assert!(rested >= 5, "replay must exercise resting orders");
7927 assert!(trades >= 50, "replay must exercise trade interleavings");
7928 }
7929}