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