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