pub mod config;
pub mod ids_generator;
pub mod inflight;
pub mod orders;
mod settlement;
#[cfg(test)]
mod tests;
use std::{
cell::RefCell,
cmp::min,
fmt::Debug,
mem,
ops::{Add, Sub},
rc::Rc,
};
use indexmap::{IndexMap, IndexSet};
use jiff::SignedDuration;
use nautilus_common::{
cache::Cache,
clock::Clock,
messages::execution::{
BatchCancelOrders, BatchModifyOrders, CancelAllOrders, CancelOrder, ModifyOrder,
},
msgbus::{self, MessagingSwitchboard},
};
use nautilus_core::{UUID4, UnixNanos, correctness::CorrectnessResult};
use nautilus_model::{
data::{
Bar, BarType, InstrumentClose, MarkPriceUpdate, OrderBookDelta, OrderBookDeltas,
OrderBookDepth, QuoteTick, TradeTick,
order::{BookOrder, OrderId},
},
enums::{
AccountType, AggregationSource, AggressorSide, BookAction, BookType, ContingencyType,
InstrumentCloseType, LiquiditySide, MarketStatus, MarketStatusAction, OmsType, OrderSide,
OrderStatus, OrderType, PositionSide, PriceType, RecordFlag, TimeInForce, TriggerType,
},
events::{
OrderAccepted, OrderCancelRejected, OrderCanceled, OrderEventAny, OrderExpired,
OrderFilled, OrderModifyRejected, OrderRejected, OrderSubmitted, OrderTriggered,
OrderUpdated,
},
identifiers::{
AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId, TraderId, Venue,
VenueOrderId,
},
instruments::{Instrument, InstrumentAny},
orderbook::{BookLevel, OrderBook},
orders::{MarketOrder, Order, OrderAny, OrderCore},
position::{Position, PositionReplayEvent},
types::{
Currency, Money, Price, Quantity,
fixed::{FIXED_PRECISION, raw_scales_match},
price::PriceRaw,
quantity::QuantityRaw,
},
};
use rust_decimal::{Decimal, RoundingStrategy};
use ustr::Ustr;
use self::{
config::OrderMatchingEngineConfig,
ids_generator::IdsGenerator,
inflight::InflightOrders,
orders::{MatchingOrders, OtoCommand},
};
use crate::{
matching::{MatchAction, OrderMatchingCore, RestingOrder},
models::{
fee::{FeeModel, FeeModelHandle},
fill::{FillModel, FillModelHandle, is_price_tradable},
},
protection::protection_price_calculate,
trailing::{store_trailing_stop_activation, trailing_stop_calculate},
};
pub struct OrderMatchingEngine {
pub venue: Venue,
pub instrument: InstrumentAny,
pub raw_id: u32,
pub book_type: BookType,
pub oms_type: OmsType,
pub account_type: AccountType,
pub market_status: MarketStatus,
pub config: OrderMatchingEngineConfig,
core: OrderMatchingCore,
clock: Rc<RefCell<dyn Clock>>,
cache: Rc<RefCell<Cache>>,
book: OrderBook,
fill_model: FillModelHandle,
fee_model: FeeModelHandle,
event_handler: Option<Rc<dyn Fn(OrderEventAny)>>,
inflight_orders: InflightOrders,
target_bid: Option<Price>,
target_ask: Option<Price>,
target_last: Option<Price>,
last_bar_bid: Option<Bar>,
last_bar_ask: Option<Bar>,
fill_at_market: bool,
collapse_bar_trades: bool,
last_bar_trade: Option<(Price, Quantity, AggressorSide)>,
execution_bar_types: IndexMap<InstrumentId, BarType>,
execution_bar_deltas: IndexMap<BarType, SignedDuration>,
account_ids: IndexMap<TraderId, AccountId>,
pending_fills: IndexMap<TradeId, PendingFill>,
post_match_order_ids: IndexSet<ClientOrderId>,
orders: MatchingOrders,
ids_generator: IdsGenerator,
last_trade_size: Option<Quantity>,
l1_trade_size: Option<QuantityRaw>,
trade_consumption: QuantityRaw,
bid_consumption: IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
ask_consumption: IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
queue_pending: IndexMap<ClientOrderId, PriceRaw>,
queue_ahead_orders: IndexMap<ClientOrderId, IndexMap<OrderId, QuantityRaw>>,
queue_ahead_total: IndexMap<ClientOrderId, (PriceRaw, QuantityRaw)>,
queue_snapshot_in_progress: bool,
queue_ids_by_price: IndexMap<PriceRaw, IndexSet<ClientOrderId>>,
queue_excess: IndexMap<ClientOrderId, QuantityRaw>,
queue_id_scratch: Vec<ClientOrderId>,
queue_pending_scratch: Vec<(ClientOrderId, PriceRaw)>,
queue_stale_scratch: Vec<ClientOrderId>,
queue_entry_scratch: Vec<(ClientOrderId, QuantityRaw, QuantityRaw)>,
prev_bid_price_raw: PriceRaw,
prev_ask_price_raw: PriceRaw,
tob_initialized: bool,
last_quote_bid: Option<Price>,
last_quote_ask: Option<Price>,
precision_mismatch_streak: u32,
instrument_close: Option<InstrumentClose>,
pending_resolution: bool,
expiration_processed: bool,
option_settlement_failed: bool,
option_settlement_warning: Option<&'static str>,
}
impl Debug for OrderMatchingEngine {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct(stringify!(OrderMatchingEngine))
.field("venue", &self.venue)
.field("instrument", &self.instrument.id())
.finish()
}
}
impl OrderMatchingEngine {
#[expect(clippy::too_many_arguments)]
pub fn new(
instrument: InstrumentAny,
raw_id: u32,
fill_model: FillModelHandle,
fee_model: FeeModelHandle,
book_type: BookType,
oms_type: OmsType,
account_type: AccountType,
clock: Rc<RefCell<dyn Clock>>,
cache: Rc<RefCell<Cache>>,
config: OrderMatchingEngineConfig,
) -> Self {
let book = OrderBook::new(instrument.id(), book_type);
let mut core = OrderMatchingCore::new(instrument.id(), instrument.price_increment());
core.set_fill_limit_inside_spread(Self::fill_limit_inside_spread_or_false(&fill_model));
let ids_generator = IdsGenerator::new(
instrument.id().venue,
oms_type,
raw_id,
config.use_random_ids,
config.use_position_ids,
cache.clone(),
);
Self {
venue: instrument.id().venue,
instrument,
raw_id,
fill_model,
fee_model,
event_handler: None,
inflight_orders: InflightOrders::default(),
book_type,
oms_type,
account_type,
clock,
cache,
book,
market_status: MarketStatus::Open,
config,
core,
target_bid: None,
target_ask: None,
target_last: None,
last_bar_bid: None,
last_bar_ask: None,
fill_at_market: true,
collapse_bar_trades: false,
last_bar_trade: None,
execution_bar_types: IndexMap::new(),
execution_bar_deltas: IndexMap::new(),
account_ids: IndexMap::new(),
pending_fills: IndexMap::new(),
post_match_order_ids: IndexSet::new(),
orders: MatchingOrders::default(),
ids_generator,
last_trade_size: None,
l1_trade_size: None,
trade_consumption: 0,
bid_consumption: IndexMap::new(),
ask_consumption: IndexMap::new(),
queue_pending: IndexMap::new(),
queue_ahead_orders: IndexMap::new(),
queue_ahead_total: IndexMap::new(),
queue_snapshot_in_progress: false,
queue_ids_by_price: IndexMap::new(),
queue_excess: IndexMap::new(),
queue_id_scratch: Vec::new(),
queue_pending_scratch: Vec::new(),
queue_stale_scratch: Vec::new(),
queue_entry_scratch: Vec::new(),
prev_bid_price_raw: 0,
prev_ask_price_raw: 0,
tob_initialized: false,
last_quote_bid: None,
last_quote_ask: None,
precision_mismatch_streak: 0,
instrument_close: None,
pending_resolution: false,
expiration_processed: false,
option_settlement_failed: false,
option_settlement_warning: None,
}
}
pub fn set_event_handler(&mut self, handler: Rc<dyn Fn(OrderEventAny)>) {
self.event_handler = Some(handler);
}
pub fn set_inflight_orders(&mut self, orders: InflightOrders) {
self.inflight_orders = orders;
}
pub fn set_matching_orders(&mut self, orders: MatchingOrders) {
self.orders = orders;
}
fn dispatch_order_event(&self, event: OrderEventAny) {
if let Some(handler) = &self.event_handler {
handler(event);
} else {
let endpoint = MessagingSwitchboard::exec_engine_process();
msgbus::send_order_event(endpoint, event);
}
}
pub fn reset(&mut self) {
self.book.reset();
self.execution_bar_types.clear();
self.execution_bar_deltas.clear();
self.account_ids.clear();
self.pending_fills.clear();
self.post_match_order_ids.clear();
self.orders
.reset(self.instrument.id(), &self.cache.borrow());
self.core.reset();
self.target_bid = None;
self.target_ask = None;
self.target_last = None;
self.last_trade_size = None;
self.l1_trade_size = None;
self.trade_consumption = 0;
self.bid_consumption.clear();
self.ask_consumption.clear();
self.queue_pending.clear();
self.queue_ahead_orders.clear();
self.queue_ahead_total.clear();
self.queue_snapshot_in_progress = false;
self.queue_ids_by_price.clear();
self.queue_excess.clear();
self.queue_id_scratch.clear();
self.queue_pending_scratch.clear();
self.queue_stale_scratch.clear();
self.queue_entry_scratch.clear();
self.prev_bid_price_raw = 0;
self.prev_ask_price_raw = 0;
self.tob_initialized = false;
self.last_quote_bid = None;
self.last_quote_ask = None;
self.last_bar_bid = None;
self.last_bar_ask = None;
self.precision_mismatch_streak = 0;
self.instrument_close = None;
self.market_status = MarketStatus::Open;
self.pending_resolution = false;
self.expiration_processed = false;
self.option_settlement_failed = false;
self.option_settlement_warning = None;
self.fill_at_market = true;
self.ids_generator.reset();
log::info!("Reset {}", self.instrument.id());
}
fn apply_liquidity_consumption(
&mut self,
mut fills: Vec<(Price, Quantity)>,
order: &OrderAny,
leaves_qty: Quantity,
book_prices: Option<&[Price]>,
trade_size: Option<QuantityRaw>,
mut journal: Option<&mut ConsumptionJournal>,
) -> Vec<(Price, Quantity)> {
if !self.config.liquidity_consumption {
return fills;
}
let mut adjusted_len = 0;
let mut remaining_qty = trade_size.map_or(leaves_qty.raw(), |trade_size| {
leaves_qty
.raw()
.min(trade_size.saturating_sub(self.trade_consumption))
});
for fill_idx in 0..fills.len() {
if remaining_qty == 0 {
break;
}
let (price, qty) = fills[fill_idx];
let book_price = book_prices
.and_then(|bp| bp.get(fill_idx).copied())
.unwrap_or(price);
let adjusted_qty_raw = self.consume_book_level(
order.order_side(),
book_price,
qty,
remaining_qty,
journal.as_deref_mut(),
);
if adjusted_qty_raw == 0 {
continue;
}
remaining_qty -= adjusted_qty_raw;
if trade_size.is_some() {
self.trade_consumption += adjusted_qty_raw;
}
let adjusted_qty = Quantity::from_raw(adjusted_qty_raw, qty.precision);
fills[adjusted_len] = (price, adjusted_qty);
adjusted_len += 1;
if order.order_type() == OrderType::MarketToLimit && order.filled_qty().is_zero() {
break;
}
}
fills.truncate(adjusted_len);
fills
}
fn consume_book_level(
&mut self,
order_side: OrderSide,
book_price: Price,
qty: Quantity,
remaining_qty: QuantityRaw,
journal: Option<&mut ConsumptionJournal>,
) -> QuantityRaw {
let book_price_raw = book_price.raw();
let level_size = self
.book
.get_quantity_at_level(book_price, order_side, qty.precision);
let consumption = match order_side {
OrderSide::Buy => &mut self.ask_consumption,
OrderSide::Sell => &mut self.bid_consumption,
};
if let Some(journal) = journal {
journal
.levels
.push((book_price_raw, consumption.get(&book_price_raw).copied()));
}
let (original_size, consumed) = consumption
.entry(book_price_raw)
.or_insert((level_size.raw(), 0));
if *original_size != level_size.raw() {
*original_size = level_size.raw();
*consumed = 0;
}
let available = original_size.saturating_sub(*consumed);
let adjusted_qty_raw = min(min(qty.raw(), available), remaining_qty);
*consumed += adjusted_qty_raw;
adjusted_qty_raw
}
fn fok_consumption_journal(&self, order: &OrderAny) -> Option<ConsumptionJournal> {
if !self.config.liquidity_consumption || order.time_in_force() != TimeInForce::Fok {
return None;
}
Some(ConsumptionJournal {
order_side: order.order_side(),
trade_consumption: self.trade_consumption,
levels: Vec::new(),
})
}
fn revert_consumption_journal(&mut self, journal: ConsumptionJournal) {
self.trade_consumption = journal.trade_consumption;
let consumption = match journal.order_side {
OrderSide::Buy => &mut self.ask_consumption,
OrderSide::Sell => &mut self.bid_consumption,
};
for (price_raw, prior) in journal.levels.into_iter().rev() {
match prior {
Some(prior) => {
consumption.insert(price_raw, prior);
}
None => {
consumption.shift_remove(&price_raw);
}
}
}
}
fn is_fok_unfillable(
&self,
order: &OrderAny,
fills: &[(Price, Quantity)],
required_qty: Quantity,
) -> bool {
if order.time_in_force() != TimeInForce::Fok {
return false;
}
let mut total_size = Quantity::zero(order.quantity().precision);
let initial_market_to_limit =
order.order_type() == OrderType::MarketToLimit && order.filled_qty().is_zero();
let mut first_price = None;
for &(fill_px, fill_qty) in fills {
if let Some(fill_px) = self.normalize_price_for_current_instrument(fill_px)
&& let Some(fill_qty) = self.normalize_quantity_for_current_instrument(fill_qty)
{
if initial_market_to_limit {
if fill_qty.is_zero() {
continue;
}
if first_price.is_some_and(|price| price != fill_px) {
break;
}
first_price = Some(fill_px);
}
total_size = total_size.add(fill_qty);
}
}
required_qty > total_size
}
fn seed_trade_consumption(
&mut self,
trade_price_raw: PriceRaw,
trade_size_raw: QuantityRaw,
trade_ts_event: UnixNanos,
aggressor_side: AggressorSide,
) {
if trade_size_raw == 0 {
return;
}
if self.book.ts_last > trade_ts_event {
return;
}
let book = &self.book;
let consumption = match aggressor_side {
AggressorSide::Buy => &mut self.ask_consumption,
AggressorSide::Sell => &mut self.bid_consumption,
AggressorSide::NoAggressor => return,
};
let mut remaining = trade_size_raw;
match aggressor_side {
AggressorSide::Buy => {
for level in book
.asks(None)
.take_while(|level| level.price.value.raw() <= trade_price_raw)
{
Self::consume_trade_level(consumption, &mut remaining, level);
if remaining == 0 {
break;
}
}
}
AggressorSide::Sell => {
for level in book
.bids(None)
.take_while(|level| level.price.value.raw() >= trade_price_raw)
{
Self::consume_trade_level(consumption, &mut remaining, level);
if remaining == 0 {
break;
}
}
}
AggressorSide::NoAggressor => unreachable!(),
}
}
fn consume_trade_level(
consumption: &mut IndexMap<PriceRaw, (QuantityRaw, QuantityRaw)>,
remaining: &mut QuantityRaw,
level: &BookLevel,
) {
let level_size = level.size_raw();
let entry = consumption
.entry(level.price.value.raw())
.or_insert((level_size, 0));
if entry.0 != level_size {
entry.0 = level_size;
entry.1 = 0;
}
let available = level_size.saturating_sub(entry.1);
let consume = min(*remaining, available);
entry.1 += consume;
*remaining -= consume;
}
pub fn set_fill_model(&mut self, fill_model: FillModelHandle) {
self.core
.set_fill_limit_inside_spread(Self::fill_limit_inside_spread_or_false(&fill_model));
self.fill_model = fill_model;
}
fn fill_limit_inside_spread_or_false(fill_model: &FillModelHandle) -> bool {
fill_model.fill_limit_inside_spread().unwrap_or_else(|e| {
log::error!("Failed to query fill model spread behavior: {e}");
false
})
}
fn snapshot_queue_position(&mut self, order: &OrderAny, price: Price) {
if !self.config.queue_position {
return;
}
let size_prec = self.instrument.size_precision();
let qty_ahead = self.book.get_quantity_at_level(
price,
OrderCore::opposite_side(order.order_side()),
size_prec,
);
let client_order_id = order.client_order_id();
self.remove_queue_position(client_order_id);
self.queue_ids_by_price
.entry(price.raw())
.or_default()
.insert(client_order_id);
if self.book_type == BookType::L1_MBP && qty_ahead.is_zero() {
let behind_bbo = match order.order_side() {
OrderSide::Buy => self.book.best_bid_price().is_some_and(|bid| price < bid),
OrderSide::Sell => self.book.best_ask_price().is_some_and(|ask| price > ask),
};
if behind_bbo {
self.queue_pending.insert(client_order_id, price.raw());
return;
}
}
self.queue_ahead_total
.insert(client_order_id, (price.raw(), qty_ahead.raw()));
if self.book_type == BookType::L3_MBO {
let orders_ahead: IndexMap<OrderId, QuantityRaw> = self
.book
.get_orders_at_level(price, OrderCore::opposite_side(order.order_side()))
.iter()
.map(|book_order| (book_order.order_id, book_order.size.raw()))
.collect();
self.queue_ahead_orders
.insert(client_order_id, orders_ahead);
}
}
fn remove_queue_position(&mut self, client_order_id: ClientOrderId) {
let pending_price = self.queue_pending.shift_remove(&client_order_id);
let ahead_price = self
.queue_ahead_total
.shift_remove(&client_order_id)
.map(|(price_raw, _)| price_raw);
self.queue_ahead_orders.shift_remove(&client_order_id);
self.queue_excess.shift_remove(&client_order_id);
for price_raw in [pending_price, ahead_price].into_iter().flatten() {
let remove_price = self
.queue_ids_by_price
.get_mut(&price_raw)
.is_some_and(|ids| {
ids.shift_remove(&client_order_id);
ids.is_empty()
});
if remove_price {
self.queue_ids_by_price.shift_remove(&price_raw);
}
}
}
fn take_queue_ids_at_price(&mut self, price_raw: PriceRaw) -> Vec<ClientOrderId> {
let mut ids = Self::take_cleared(&mut self.queue_id_scratch);
if let Some(tracked_ids) = self.queue_ids_by_price.get(&price_raw) {
ids.extend(tracked_ids.iter().copied());
}
ids
}
fn decrement_queue_on_trade(
&mut self,
price_raw: PriceRaw,
trade_size_raw: QuantityRaw,
aggressor_side: AggressorSide,
) {
if !self.config.queue_position {
return;
}
self.queue_excess.clear();
let keys = self.take_queue_ids_at_price(price_raw);
let mut entries = Self::take_cleared(&mut self.queue_entry_scratch);
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
for client_order_id in keys.iter().copied() {
let (order_price_raw, ahead_raw) =
match self.queue_ahead_total.get(&client_order_id).copied() {
Some(v) => v,
None => continue,
};
let cache = self.cache.borrow();
let order_info = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
return None;
}
let has_pending_updates = self
.orders
.pending_updates
.borrow()
.contains_key(&client_order_id);
let has_pending_fills = self
.orders
.cached_filled_qty
.borrow()
.get(&client_order_id)
.is_some_and(|filled_qty| *filled_qty != order.filled_qty());
let snapshot;
let order = if has_pending_updates || has_pending_fills {
snapshot = self.order_snapshot(client_order_id)?;
&snapshot
} else {
&order
};
Some((order.order_side(), order.leaves_qty().raw()))
});
drop(cache);
let Some((order_side, leaves_raw)) = order_info else {
stale.push(client_order_id);
continue;
};
if order_price_raw != price_raw || ahead_raw == 0 {
continue;
}
let should_decrement = matches!(aggressor_side, AggressorSide::NoAggressor)
|| (aggressor_side == AggressorSide::Buy && order_side == OrderSide::Sell)
|| (aggressor_side == AggressorSide::Sell && order_side == OrderSide::Buy);
if should_decrement {
entries.push((client_order_id, ahead_raw, leaves_raw));
}
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
entries.sort_by_key(|&(_, ahead, _)| ahead);
let mut remaining = trade_size_raw;
let mut prev_position: QuantityRaw = 0;
for (client_order_id, ahead_raw, leaves_raw) in &entries {
if remaining == 0 {
let new_ahead = ahead_raw.saturating_sub(trade_size_raw);
self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, new_ahead);
if new_ahead == 0 {
self.queue_excess.insert(*client_order_id, 0);
}
continue;
}
let gap = ahead_raw.saturating_sub(prev_position);
let queue_consumed = remaining.min(gap);
remaining -= queue_consumed;
if remaining == 0 && queue_consumed < gap {
let new_ahead = ahead_raw.saturating_sub(trade_size_raw);
self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, new_ahead);
continue;
}
self.reduce_queue_ahead(*client_order_id, price_raw, *ahead_raw, 0);
let excess = remaining.min(*leaves_raw);
self.queue_excess.insert(*client_order_id, excess);
remaining -= excess;
prev_position = ahead_raw + excess;
}
self.queue_id_scratch = keys;
self.queue_entry_scratch = entries;
self.queue_stale_scratch = stale;
}
fn reduce_queue_ahead(
&mut self,
client_order_id: ClientOrderId,
price_raw: PriceRaw,
ahead_raw: QuantityRaw,
new_ahead_raw: QuantityRaw,
) {
self.queue_ahead_total
.insert(client_order_id, (price_raw, new_ahead_raw));
self.consume_queue_ahead_orders(client_order_id, ahead_raw.saturating_sub(new_ahead_raw));
}
fn consume_queue_ahead_orders(
&mut self,
client_order_id: ClientOrderId,
mut amount_raw: QuantityRaw,
) {
let Some(orders_ahead) = self.queue_ahead_orders.get_mut(&client_order_id) else {
return;
};
while amount_raw > 0 {
let Some((&book_order_id, &size_raw)) = orders_ahead.get_index(0) else {
break;
};
if size_raw <= amount_raw {
orders_ahead.shift_remove(&book_order_id);
amount_raw -= size_raw;
} else {
orders_ahead.insert(book_order_id, size_raw - amount_raw);
amount_raw = 0;
}
}
}
fn determine_trade_fill_qty(&self, order: &OrderAny) -> Option<QuantityRaw> {
if !self.config.queue_position {
return Some(order.leaves_qty().raw());
}
let client_order_id = order.client_order_id();
if self.queue_pending.contains_key(&client_order_id) {
return None;
}
if let Some(&(tracked_price_raw, ahead_raw)) = self.queue_ahead_total.get(&client_order_id)
&& let Some(order_price) = order.price()
&& order_price.raw() == tracked_price_raw
&& ahead_raw > 0
{
return None;
}
let leaves_raw = order.leaves_qty().raw();
if leaves_raw == 0 {
return None;
}
let mut available_raw = leaves_raw;
if let Some(trade_size) = self.last_trade_size {
let remaining = trade_size.raw().saturating_sub(self.trade_consumption);
available_raw = available_raw.min(remaining);
if let Some(&excess_raw) = self.queue_excess.get(&client_order_id) {
if excess_raw == 0 {
return None;
}
available_raw = available_raw.min(excess_raw);
}
}
if available_raw == 0 {
return None;
}
Some(available_raw)
}
fn rebase_queue_positions(&mut self) {
if !self.config.queue_position {
return;
}
let tracked: Vec<_> = self
.queue_ahead_total
.iter()
.map(|(&client_order_id, &(price_raw, ahead_raw))| {
(client_order_id, price_raw, ahead_raw)
})
.collect();
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
let size_precision = self.instrument.size_precision();
let price_precision = self.instrument.price_precision();
for (client_order_id, price_raw, ahead_raw) in tracked {
let order_side = self
.cache
.borrow()
.order(&client_order_id)
.and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
let Some(order_side) = order_side else {
stale.push(client_order_id);
continue;
};
let price = Price::from_raw(price_raw, price_precision);
let visible_raw = self
.book
.get_quantity_at_level(price, OrderCore::opposite_side(order_side), size_precision)
.raw();
let rebased_raw = ahead_raw.min(visible_raw);
if self.book_type == BookType::L3_MBO {
let previous_orders = self
.queue_ahead_orders
.get(&client_order_id)
.cloned()
.unwrap_or_default();
let mut orders_ahead = IndexMap::new();
let mut total_raw = 0;
for book_order in self
.book
.get_orders_at_level(price, OrderCore::opposite_side(order_side))
{
if !previous_orders.contains_key(&book_order.order_id) {
continue;
}
let previous_size_raw = previous_orders[&book_order.order_id];
let size_raw = previous_size_raw.min(book_order.size.raw());
orders_ahead.insert(book_order.order_id, size_raw);
total_raw += size_raw;
}
self.queue_ahead_orders
.insert(client_order_id, orders_ahead);
self.queue_ahead_total
.insert(client_order_id, (price_raw, total_raw));
} else {
self.queue_ahead_total
.insert(client_order_id, (price_raw, rebased_raw));
}
}
for client_order_id in stale.drain(..) {
self.remove_queue_position(client_order_id);
}
self.queue_stale_scratch = stale;
}
fn adjust_queue_for_delta(&mut self, delta: &OrderBookDelta) {
if delta.action == BookAction::Delete {
if self.is_order_granular_delta(delta.flags) {
self.advance_l3_queue_on_delete(delta.order.order_id);
} else {
self.clear_queue_on_delete(delta.order.price.raw(), delta.order.side);
}
} else if delta.action == BookAction::Update {
if self.is_order_granular_delta(delta.flags) {
self.adjust_l3_queue_on_update(&delta.order);
} else {
self.cap_queue_ahead(
delta.order.price.raw(),
delta.order.size.raw(),
delta.order.side,
);
}
}
}
fn clear_queue_on_delete(
&mut self,
deleted_price_raw: PriceRaw,
deleted_side: Option<OrderSide>,
) {
let keys = self.take_queue_ids_at_price(deleted_price_raw);
for client_order_id in keys.iter().copied() {
if let Some(&(order_price_raw, ahead_raw)) =
self.queue_ahead_total.get(&client_order_id)
&& order_price_raw == deleted_price_raw
{
let matches_side = self
.cache
.borrow()
.order(&client_order_id)
.is_some_and(|o| Some(o.order_side()) == deleted_side);
if matches_side {
self.reduce_queue_ahead(client_order_id, order_price_raw, ahead_raw, 0);
}
}
}
self.queue_id_scratch = keys;
}
fn is_order_granular_delta(&self, flags: u8) -> bool {
self.book_type == BookType::L3_MBO
&& !RecordFlag::F_TOB.matches(flags)
&& !RecordFlag::F_MBP.matches(flags)
}
fn advance_l3_queue_on_delete(&mut self, book_order_id: OrderId) {
for (client_order_id, orders_ahead) in &mut self.queue_ahead_orders {
let Some(size_raw) = orders_ahead.shift_remove(&book_order_id) else {
continue;
};
if let Some((_, ahead_raw)) = self.queue_ahead_total.get_mut(client_order_id) {
*ahead_raw = ahead_raw.saturating_sub(size_raw);
}
}
}
fn adjust_l3_queue_on_update(&mut self, book_order: &BookOrder) {
for (client_order_id, orders_ahead) in &mut self.queue_ahead_orders {
let Some(&tracked_size_raw) = orders_ahead.get(&book_order.order_id) else {
continue;
};
let Some((tracked_price_raw, ahead_raw)) =
self.queue_ahead_total.get_mut(client_order_id)
else {
continue;
};
if book_order.price.raw() != *tracked_price_raw {
*ahead_raw = ahead_raw.saturating_sub(tracked_size_raw);
orders_ahead.shift_remove(&book_order.order_id);
} else if book_order.size.raw() < tracked_size_raw {
*ahead_raw = ahead_raw.saturating_sub(tracked_size_raw - book_order.size.raw());
orders_ahead.insert(book_order.order_id, book_order.size.raw());
} else if book_order.size.raw() > tracked_size_raw {
*ahead_raw = ahead_raw.saturating_add(book_order.size.raw() - tracked_size_raw);
orders_ahead.insert(book_order.order_id, book_order.size.raw());
}
}
}
fn cap_queue_ahead(
&mut self,
price_raw: PriceRaw,
size_raw: QuantityRaw,
order_side: Option<OrderSide>,
) {
let keys = self.take_queue_ids_at_price(price_raw);
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
for client_order_id in keys.iter().copied() {
let (order_price_raw, ahead_raw) =
match self.queue_ahead_total.get(&client_order_id).copied() {
Some(v) => v,
None => continue,
};
if order_price_raw != price_raw || ahead_raw <= size_raw {
continue;
}
let cache = self.cache.borrow();
let order_info = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
drop(cache);
let Some(side) = order_info else {
stale.push(client_order_id);
continue;
};
if Some(side) != order_side {
continue;
}
self.reduce_queue_ahead(client_order_id, order_price_raw, ahead_raw, size_raw);
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
self.queue_id_scratch = keys;
self.queue_stale_scratch = stale;
}
fn seed_tob_baseline(&mut self) {
let bid = self.book.best_bid_price();
let ask = self.book.best_ask_price();
self.prev_bid_price_raw = bid.map_or(0, |p| p.raw());
self.prev_ask_price_raw = ask.map_or(0, |p| p.raw());
self.tob_initialized = bid.is_some() || ask.is_some();
}
fn decrement_l1_queue_on_quote(
&mut self,
bid_price_raw: PriceRaw,
bid_size_raw: QuantityRaw,
ask_price_raw: PriceRaw,
ask_size_raw: QuantityRaw,
) {
if !self.config.queue_position {
return;
}
if self.tob_initialized {
if bid_price_raw < self.prev_bid_price_raw {
self.adjust_l1_queue_on_price_move(bid_price_raw, bid_size_raw, OrderSide::Buy);
}
if ask_price_raw > self.prev_ask_price_raw {
self.adjust_l1_queue_on_price_move(ask_price_raw, ask_size_raw, OrderSide::Sell);
}
}
self.resolve_pending_l1_snapshots(bid_price_raw, bid_size_raw, ask_price_raw, ask_size_raw);
self.cap_queue_ahead(bid_price_raw, bid_size_raw, Some(OrderSide::Buy));
self.cap_queue_ahead(ask_price_raw, ask_size_raw, Some(OrderSide::Sell));
}
fn adjust_l1_queue_on_price_move(
&mut self,
new_price_raw: PriceRaw,
new_size_raw: QuantityRaw,
order_side: OrderSide,
) {
let mut keys = Self::take_cleared(&mut self.queue_id_scratch);
keys.extend(self.queue_ahead_total.keys().copied());
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
for client_order_id in keys.iter().copied() {
let Some(&(order_price_raw, ahead_raw)) = self.queue_ahead_total.get(&client_order_id)
else {
continue;
};
let cache = self.cache.borrow();
let order_info = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
drop(cache);
let Some(side) = order_info else {
stale.push(client_order_id);
continue;
};
if side != order_side {
continue;
}
let crossed = match order_side {
OrderSide::Buy => order_price_raw > new_price_raw,
_ => order_price_raw < new_price_raw,
};
if crossed {
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, 0));
} else if order_price_raw == new_price_raw && ahead_raw > new_size_raw {
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, new_size_raw));
}
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
let mut pending = Self::take_cleared(&mut self.queue_pending_scratch);
pending.extend(
self.queue_pending
.iter()
.map(|(&client_order_id, &price_raw)| (client_order_id, price_raw)),
);
for (client_order_id, order_price_raw) in pending.iter().copied() {
let cache = self.cache.borrow();
let order_info = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
drop(cache);
let Some(side) = order_info else {
stale.push(client_order_id);
continue;
};
if side != order_side {
continue;
}
let crossed = match order_side {
OrderSide::Buy => order_price_raw > new_price_raw,
_ => order_price_raw < new_price_raw,
};
if crossed {
self.queue_pending.shift_remove(&client_order_id);
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, 0));
} else if order_price_raw == new_price_raw {
self.queue_pending.shift_remove(&client_order_id);
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, new_size_raw));
}
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
self.queue_id_scratch = keys;
self.queue_pending_scratch = pending;
self.queue_stale_scratch = stale;
}
fn resolve_pending_l1_snapshots(
&mut self,
bid_price_raw: PriceRaw,
bid_size_raw: QuantityRaw,
ask_price_raw: PriceRaw,
ask_size_raw: QuantityRaw,
) {
let mut keys = self.take_queue_ids_at_price(bid_price_raw);
if ask_price_raw != bid_price_raw
&& let Some(ask_ids) = self.queue_ids_by_price.get(&ask_price_raw)
{
keys.extend(ask_ids.iter().copied());
}
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
for client_order_id in keys.iter().copied() {
let Some(&order_price_raw) = self.queue_pending.get(&client_order_id) else {
continue;
};
let cache = self.cache.borrow();
let order_info = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
drop(cache);
let Some(side) = order_info else {
stale.push(client_order_id);
continue;
};
let matched_size = match side {
OrderSide::Buy if order_price_raw == bid_price_raw => Some(bid_size_raw),
OrderSide::Sell if order_price_raw == ask_price_raw => Some(ask_size_raw),
_ => None,
};
if let Some(size) = matched_size {
self.queue_pending.shift_remove(&client_order_id);
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, size));
}
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
self.queue_id_scratch = keys;
self.queue_stale_scratch = stale;
}
fn resolve_pending_on_trade(&mut self, trade_price_raw: PriceRaw) {
let mut pending = Self::take_cleared(&mut self.queue_pending_scratch);
pending.extend(
self.queue_pending
.iter()
.map(|(&client_order_id, &price_raw)| (client_order_id, price_raw)),
);
let mut stale = Self::take_cleared(&mut self.queue_stale_scratch);
for (client_order_id, order_price_raw) in pending.iter().copied() {
let cache = self.cache.borrow();
let order_side = cache.order(&client_order_id).and_then(|order| {
if order.is_closed() {
None
} else {
Some(order.order_side())
}
});
drop(cache);
let Some(side) = order_side else {
stale.push(client_order_id);
continue;
};
let crossed = match side {
OrderSide::Buy => trade_price_raw < order_price_raw,
OrderSide::Sell => trade_price_raw > order_price_raw,
};
if crossed {
self.queue_pending.shift_remove(&client_order_id);
self.queue_ahead_total
.insert(client_order_id, (order_price_raw, 0));
}
}
for id in stale.drain(..) {
self.remove_queue_position(id);
}
self.queue_pending_scratch = pending;
self.queue_stale_scratch = stale;
}
fn take_cleared<T>(buf: &mut Vec<T>) -> Vec<T> {
let mut items = mem::take(buf);
items.clear();
items
}
#[must_use]
pub fn best_bid_price(&self) -> Option<Price> {
self.book.best_bid_price()
}
#[must_use]
pub fn best_ask_price(&self) -> Option<Price> {
self.book.best_ask_price()
}
#[must_use]
pub const fn get_book(&self) -> &OrderBook {
&self.book
}
#[must_use]
pub fn get_open_bid_orders(&self) -> Vec<RestingOrder> {
self.core.get_orders_bid()
}
#[must_use]
pub fn get_open_ask_orders(&self) -> Vec<RestingOrder> {
self.core.get_orders_ask()
}
#[must_use]
pub fn get_open_orders(&self) -> Vec<RestingOrder> {
self.core.get_orders()
}
#[must_use]
pub fn order_exists(&self, client_order_id: ClientOrderId) -> bool {
self.core.order_exists(client_order_id)
}
#[must_use]
pub fn cached_filled_qty_len(&self) -> usize {
let cache = self.cache.borrow();
self.orders
.cached_filled_qty
.borrow()
.keys()
.filter(|id| {
cache
.order(id)
.is_some_and(|order| order.instrument_id() == self.instrument.id())
})
.count()
}
#[must_use]
pub const fn get_core(&self) -> &OrderMatchingCore {
&self.core
}
pub fn set_fill_at_market(&mut self, value: bool) {
self.fill_at_market = value;
}
pub fn update_instrument(&mut self, instrument: InstrumentAny) -> anyhow::Result<()> {
if instrument.id() != self.instrument.id() {
anyhow::bail!(
"Cannot update instrument {} with {}",
self.instrument.id(),
instrument.id()
);
}
let changed = instrument.price_increment() != self.instrument.price_increment()
|| instrument.price_precision() != self.instrument.price_precision()
|| instrument.size_precision() != self.instrument.size_precision();
if changed {
self.core
.update_price_increment(instrument.price_increment());
self.book.reset();
self.l1_trade_size = None;
self.trade_consumption = 0;
self.bid_consumption.clear();
self.ask_consumption.clear();
self.queue_pending.clear();
self.queue_ahead_orders.clear();
self.queue_ahead_total.clear();
self.queue_ids_by_price.clear();
self.queue_excess.clear();
self.prev_bid_price_raw = 0;
self.prev_ask_price_raw = 0;
self.tob_initialized = false;
self.last_quote_bid = None;
self.last_quote_ask = None;
self.precision_mismatch_streak = 0;
self.target_bid = None;
self.target_ask = None;
self.target_last = None;
self.last_bar_bid = None;
self.last_bar_ask = None;
self.core.bid = None;
self.core.ask = None;
self.core.last = None;
log::info!(
"Updated instrument {} (price_precision={} size_precision={})",
instrument.id(),
instrument.price_precision(),
instrument.size_precision()
);
}
self.instrument = instrument;
if changed {
self.drop_incompatible_core_orders();
}
Ok(())
}
fn check_price_precision(&self, actual: u8, field: &str) -> anyhow::Result<()> {
let expected = self.instrument.price_precision();
if actual != expected {
anyhow::bail!(
"Invalid {field} precision {actual}, expected {expected} for {}",
self.instrument.id()
);
}
Ok(())
}
fn check_size_precision(&self, actual: u8, field: &str) -> anyhow::Result<()> {
let expected = self.instrument.size_precision();
if actual != expected {
anyhow::bail!(
"Invalid {field} precision {actual}, expected {expected} for {}",
self.instrument.id()
);
}
Ok(())
}
fn log_precision_mismatch(
&mut self,
data_type: &str,
instrument_id: InstrumentId,
err: &anyhow::Error,
) {
self.precision_mismatch_streak = self.precision_mismatch_streak.saturating_add(1);
let streak = self.precision_mismatch_streak;
if streak <= 3 || streak.is_multiple_of(100) {
log::warn!(
"Skipping {data_type} for {instrument_id}: {err} \
(consecutive_precision_mismatches={streak})"
);
}
if streak == 20 {
log::error!(
"Precision mismatches reached {streak} consecutive events for \
{instrument_id}; check instrument update flow and upstream market data"
);
}
}
fn drop_incompatible_core_orders(&mut self) {
let client_order_ids: Vec<ClientOrderId> = self
.core
.iter_orders()
.filter(|order| {
!self.resting_order_matches_current_instrument(order)
|| !self.cached_order_matches_current_instrument(order.client_order_id)
})
.map(|order| order.client_order_id)
.collect();
for client_order_id in client_order_ids {
let order = self
.cache
.borrow()
.order(&client_order_id)
.map(|o| o.clone());
if let Some(order) = order
&& (order.is_inflight() || order.is_open())
{
log::warn!(
"Canceling order {client_order_id} after instrument update: \
price, trigger price, or quantity is not compatible with {}",
self.instrument.id()
);
self.cancel_order(&order, None);
} else {
self.delete_core_order(client_order_id);
self.orders
.cached_filled_qty
.borrow_mut()
.swap_remove(&client_order_id);
}
}
}
fn cached_order_matches_current_instrument(&self, client_order_id: ClientOrderId) -> bool {
self.cache
.borrow()
.order(&client_order_id)
.is_none_or(|order| {
Self::quantity_matches_precision(order.quantity(), self.instrument.size_precision())
})
}
fn resting_order_matches_current_instrument(&self, order: &RestingOrder) -> bool {
order
.limit_price
.is_none_or(|price| self.price_matches_current_instrument(price))
&& order
.trigger_price
.is_none_or(|price| self.price_matches_current_instrument(price))
}
fn price_matches_current_instrument(&self, price: Price) -> bool {
Self::price_matches_precision(price, self.instrument.price_precision())
&& Self::price_matches_tick(price, self.instrument.price_increment())
}
fn price_matches_precision(price: Price, precision: u8) -> bool {
let precision_diff = FIXED_PRECISION.saturating_sub(precision);
let scale = PriceRaw::pow(10, u32::from(precision_diff));
price.raw() % scale == 0
}
fn price_matches_tick(price: Price, increment: Price) -> bool {
let increment_raw = increment.raw().abs();
increment_raw == 0 || price.raw() % increment_raw == 0
}
fn quantity_matches_precision(quantity: Quantity, precision: u8) -> bool {
let precision_diff = FIXED_PRECISION.saturating_sub(precision);
let scale = QuantityRaw::pow(10, u32::from(precision_diff));
quantity.raw().is_multiple_of(scale)
}
fn normalize_price_for_current_instrument(&self, price: Price) -> Option<Price> {
if !self.price_matches_current_instrument(price) {
return None;
}
Some(Price::from_raw(
price.raw(),
self.instrument.price_precision(),
))
}
fn normalize_quantity_for_current_instrument(&self, quantity: Quantity) -> Option<Quantity> {
let precision = self.instrument.size_precision();
if !Self::quantity_matches_precision(quantity, precision) {
return None;
}
Some(Quantity::from_raw(quantity.raw(), precision))
}
pub fn process_order_book_delta(&mut self, delta: &OrderBookDelta) -> anyhow::Result<()> {
log::debug!("Processing {delta}");
if matches!(delta.action, BookAction::Add | BookAction::Update) {
self.check_price_precision(delta.order.price.precision, "delta order price")?;
self.check_size_precision(delta.order.size.precision, "delta order size")?;
}
if self.book_type == BookType::L1_MBP {
self.iterate(delta.ts_init, AggressorSide::NoAggressor);
return Ok(());
}
self.book.apply_delta(delta)?;
let is_snapshot = RecordFlag::F_SNAPSHOT.matches(delta.flags);
let is_last = RecordFlag::F_LAST.matches(delta.flags);
let is_clear = delta.action == BookAction::Clear;
let snapshot_complete = is_last && (is_snapshot || self.queue_snapshot_in_progress);
if self.config.queue_position {
if is_snapshot && !is_last {
self.queue_snapshot_in_progress = true;
}
if snapshot_complete {
self.queue_snapshot_in_progress = false;
self.rebase_queue_positions();
} else if is_clear && !is_snapshot {
self.rebase_queue_positions();
} else if !self.queue_snapshot_in_progress {
self.adjust_queue_for_delta(delta);
}
}
if self.config.queue_position && (snapshot_complete || (is_clear && !is_snapshot)) {
self.seed_tob_baseline();
}
self.iterate(delta.ts_init, AggressorSide::NoAggressor);
Ok(())
}
pub fn process_order_book_deltas(&mut self, deltas: &OrderBookDeltas) -> anyhow::Result<()> {
log::debug!("Processing {deltas}");
for delta in &deltas.deltas {
if matches!(delta.action, BookAction::Add | BookAction::Update) {
self.check_price_precision(delta.order.price.precision, "delta order price")?;
self.check_size_precision(delta.order.size.precision, "delta order size")?;
}
}
if self.book_type == BookType::L1_MBP {
self.iterate(deltas.ts_init, AggressorSide::NoAggressor);
return Ok(());
}
self.book.apply_deltas(deltas)?;
let mut has_snapshot_or_clear = false;
if self.config.queue_position {
for delta in &deltas.deltas {
if RecordFlag::F_SNAPSHOT.matches(delta.flags) || delta.action == BookAction::Clear
{
has_snapshot_or_clear = true;
break;
}
self.adjust_queue_for_delta(delta);
}
}
if self.config.queue_position && has_snapshot_or_clear {
self.queue_snapshot_in_progress = false;
self.rebase_queue_positions();
self.seed_tob_baseline();
}
self.iterate(deltas.ts_init, AggressorSide::NoAggressor);
Ok(())
}
pub fn process_order_book_depth(&mut self, depth: &OrderBookDepth) -> anyhow::Result<()> {
log::debug!("Processing OrderBookDepth for {}", depth.instrument_id);
for order in &depth.bids {
if order.side.is_none() || !order.size.is_positive() {
continue;
}
self.check_price_precision(order.price.precision, "bid price")?;
self.check_size_precision(order.size.precision, "bid size")?;
}
for order in &depth.asks {
if order.side.is_none() || !order.size.is_positive() {
continue;
}
self.check_price_precision(order.price.precision, "ask price")?;
self.check_size_precision(order.size.precision, "ask size")?;
}
let top_bid = Self::first_valid_depth_order(&depth.bids, OrderSide::Buy);
let top_ask = Self::first_valid_depth_order(&depth.asks, OrderSide::Sell);
if self.book_type == BookType::L1_MBP {
let quote = QuoteTick::new(
depth.instrument_id,
Self::depth_quote_price(top_bid, self.instrument.price_precision()),
Self::depth_quote_price(top_ask, self.instrument.price_precision()),
Self::depth_quote_size(top_bid, self.instrument.size_precision()),
Self::depth_quote_size(top_ask, self.instrument.size_precision()),
depth.ts_event,
depth.ts_init,
);
self.book.update_quote_tick("e)?;
self.reset_trade_liquidity(quote.ts_event);
self.last_quote_bid = top_bid.map(|order| order.price);
self.last_quote_ask = top_ask.map(|order| order.price);
} else {
self.book.apply_depth(depth)?;
}
if self.config.queue_position {
self.rebase_queue_positions();
let bid_price_raw = top_bid.map_or(0, |order| order.price.raw());
let bid_size_raw = top_bid.map_or(0, |order| order.size.raw());
let ask_price_raw = top_ask.map_or(0, |order| order.price.raw());
let ask_size_raw = top_ask.map_or(0, |order| order.size.raw());
self.decrement_l1_queue_on_quote(
bid_price_raw,
bid_size_raw,
ask_price_raw,
ask_size_raw,
);
self.prev_bid_price_raw = bid_price_raw;
self.prev_ask_price_raw = ask_price_raw;
self.tob_initialized = true;
}
self.iterate(depth.ts_init, AggressorSide::NoAggressor);
Ok(())
}
fn first_valid_depth_order(orders: &[BookOrder], side: OrderSide) -> Option<BookOrder> {
orders
.iter()
.copied()
.find(|order| order.side == Some(side) && order.size.is_positive())
}
fn depth_quote_price(order: Option<BookOrder>, price_precision: u8) -> Price {
order.map_or_else(|| Price::zero(price_precision), |order| order.price)
}
fn depth_quote_size(order: Option<BookOrder>, size_precision: u8) -> Quantity {
order.map_or_else(|| Quantity::zero(size_precision), |order| order.size)
}
pub fn process_mark_price(&mut self, mark: &MarkPriceUpdate) {
self.core.mark = Some(mark.value);
self.iterate_with_mode(
mark.ts_init,
AggressorSide::NoAggressor,
OrderMatchMode::StopTriggers(TriggerType::MarkPrice),
);
}
pub fn process_quote_tick(&mut self, quote: &QuoteTick) {
log::debug!("Processing {quote}");
if let Err(e) = self.check_price_precision(quote.bid_price.precision, "bid_price") {
self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
return;
}
if let Err(e) = self.check_price_precision(quote.ask_price.precision, "ask_price") {
self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
return;
}
if let Err(e) = self.check_size_precision(quote.bid_size.precision, "bid_size") {
self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
return;
}
if let Err(e) = self.check_size_precision(quote.ask_size.precision, "ask_size") {
self.log_precision_mismatch("quote tick", quote.instrument_id, &e);
return;
}
self.precision_mismatch_streak = 0;
if self.book_type == BookType::L1_MBP {
if quote.ts_event < self.book.ts_last {
log::warn!(
"Skipping stale quote: ts_event {} < book.ts_last {} for {}",
quote.ts_event,
self.book.ts_last,
self.book.instrument_id,
);
self.iterate(quote.ts_init, AggressorSide::NoAggressor);
return;
}
if !self.update_quote_tick_or_skip(quote, "quote tick") {
return;
}
if self.config.queue_position {
self.decrement_l1_queue_on_quote(
quote.bid_price.raw(),
quote.bid_size.raw(),
quote.ask_price.raw(),
quote.ask_size.raw(),
);
self.prev_bid_price_raw = quote.bid_price.raw();
self.prev_ask_price_raw = quote.ask_price.raw();
self.tob_initialized = true;
}
self.last_quote_bid = Some(quote.bid_price);
self.last_quote_ask = Some(quote.ask_price);
}
self.iterate(quote.ts_init, AggressorSide::NoAggressor);
}
pub fn process_bar(&mut self, bar: &Bar) {
log::debug!("Processing {bar}");
debug_assert!(
bar.high >= bar.open
&& bar.high >= bar.low
&& bar.high >= bar.close
&& bar.low <= bar.open
&& bar.low <= bar.close,
"OHLC invariant violated for {bar}"
);
if !self.config.bar_execution || self.book_type != BookType::L1_MBP {
return;
}
let bar_type = bar.bar_type;
if bar_type.aggregation_source() == AggregationSource::Internal {
return;
}
if let Err(e) = self.check_price_precision(bar.open.precision, "bar open") {
self.log_precision_mismatch("bar", bar.instrument_id(), &e);
return;
}
if let Err(e) = self.check_price_precision(bar.high.precision, "bar high") {
self.log_precision_mismatch("bar", bar.instrument_id(), &e);
return;
}
if let Err(e) = self.check_price_precision(bar.low.precision, "bar low") {
self.log_precision_mismatch("bar", bar.instrument_id(), &e);
return;
}
if let Err(e) = self.check_price_precision(bar.close.precision, "bar close") {
self.log_precision_mismatch("bar", bar.instrument_id(), &e);
return;
}
if let Err(e) = self.check_size_precision(bar.volume.precision, "bar volume") {
self.log_precision_mismatch("bar", bar.instrument_id(), &e);
return;
}
self.precision_mismatch_streak = 0;
let price_type = bar_type.spec().price_type;
if price_type == PriceType::Mark {
log::warn!(
"Cannot process bar for {} with `PriceType::Mark`, mark price bars are not supported for bar execution",
bar.instrument_id(),
);
return;
}
let execution_bar_type =
if let Some(execution_bar_type) = self.execution_bar_types.get(&bar.instrument_id()) {
execution_bar_type.to_owned()
} else {
self.execution_bar_types
.insert(bar.instrument_id(), bar_type);
self.execution_bar_deltas
.insert(bar_type, bar_type.spec().timedelta());
bar_type
};
if execution_bar_type != bar_type {
let bar_type_timedelta = *self
.execution_bar_deltas
.entry(bar_type)
.or_insert_with(|| bar_type.spec().timedelta());
if self.execution_bar_deltas.get(&execution_bar_type).unwrap() >= &bar_type_timedelta {
self.execution_bar_types
.insert(bar_type.instrument_id(), bar_type);
} else {
return;
}
}
match price_type {
PriceType::Last | PriceType::Mid => self.process_trade_ticks_from_bar(bar),
PriceType::Bid => {
self.last_bar_bid = Some(bar.to_owned());
self.process_quote_ticks_from_bar();
}
PriceType::Ask => {
self.last_bar_ask = Some(bar.to_owned());
self.process_quote_ticks_from_bar();
}
PriceType::Mark => {
unreachable!("PriceType::Mark bars return before execution bar state updates")
}
}
}
fn process_trade_ticks_from_bar(&mut self, bar: &Bar) {
self.collapse_bar_trades = self.has_nothing_to_match(bar.ts_init);
let sizes = BarTickSizes::from_volume(bar.volume, self.instrument.size_increment());
let aggressor_side = if self.core.last.is_none_or(|last| bar.open > last) {
AggressorSide::Buy
} else {
AggressorSide::Sell
};
if sizes.open.is_positive() || self.core.last.is_none_or(|last| bar.open != last) {
self.fill_at_market = true;
if !self.process_bar_trade_tick(
bar,
bar.open,
sizes.open,
aggressor_side,
"bar open trade tick",
) {
return;
}
}
let high_first = self.bar_high_first(bar);
if high_first {
self.process_bar_high(bar, sizes.high);
self.process_bar_low(bar, sizes.low);
} else {
self.process_bar_low(bar, sizes.low);
self.process_bar_high(bar, sizes.high);
}
if sizes.close.is_positive() || self.core.last.is_some_and(|last| bar.close != last) {
let aggressor_side = if bar.close > self.core.last.unwrap() {
AggressorSide::Buy
} else {
AggressorSide::Sell
};
if !self.process_bar_trade_tick(
bar,
bar.close,
sizes.close,
aggressor_side,
"bar close trade tick",
) {
return;
}
}
self.collapse_bar_trades = false;
if let Some((price, size, aggressor_side)) = self.last_bar_trade.take() {
let last = self.core.last;
self.process_bar_trade_tick(bar, price, size, aggressor_side, "bar trade tick");
self.core.last = last;
}
self.fill_at_market = true;
}
fn process_bar_high(&mut self, bar: &Bar, size: Quantity) {
if size.is_positive() || self.core.last.is_some_and(|last| bar.high > last) {
self.process_bar_trade_tick(
bar,
bar.high,
size,
AggressorSide::Buy,
"bar high trade tick",
);
}
}
fn process_bar_low(&mut self, bar: &Bar, size: Quantity) {
if size.is_positive() || self.core.last.is_some_and(|last| bar.low < last) {
self.process_bar_trade_tick(
bar,
bar.low,
size,
AggressorSide::Sell,
"bar low trade tick",
);
}
}
fn process_bar_trade_tick(
&mut self,
bar: &Bar,
price: Price,
size: Quantity,
aggressor_side: AggressorSide,
context: &str,
) -> bool {
if price != bar.open {
self.fill_at_market = false;
}
if size.is_zero() {
self.core.set_last_raw(price);
return true;
}
if self.collapse_bar_trades {
if self.last_bar_trade.is_some() {
self.ids_generator.skip_trade_id();
}
self.last_bar_trade = Some((price, size, aggressor_side));
self.core.set_last_raw(price);
return true;
}
let trade_tick = TradeTick::new(
bar.instrument_id(),
price,
size,
aggressor_side,
self.ids_generator.generate_trade_id(bar.ts_init),
bar.ts_init,
bar.ts_init,
);
if !self.update_trade_tick_or_skip(&trade_tick, context) {
return false;
}
self.core.set_last_raw(price);
self.iterate(trade_tick.ts_init, AggressorSide::NoAggressor);
true
}
fn process_quote_ticks_from_bar(&mut self) {
if self.last_bar_bid.is_none()
|| self.last_bar_ask.is_none()
|| self.last_bar_bid.unwrap().ts_init != self.last_bar_ask.unwrap().ts_init
{
return;
}
let bid_bar = self.last_bar_bid.unwrap();
let ask_bar = self.last_bar_ask.unwrap();
let size_increment = self.instrument.size_increment();
let bid_sizes = BarTickSizes::from_volume(bid_bar.volume, size_increment);
let ask_sizes = BarTickSizes::from_volume(ask_bar.volume, size_increment);
let mut has_current_bid = false;
let mut has_current_ask = false;
let mut quote_tick = QuoteTick::new(
self.book.instrument_id,
bid_bar.open,
ask_bar.open,
bid_sizes.open,
ask_sizes.open,
bid_bar.ts_init,
bid_bar.ts_init,
);
self.fill_at_market = true;
if !self.process_bar_quote_tick(
"e_tick,
"bar open quote tick",
&mut has_current_bid,
&mut has_current_ask,
) {
return;
}
let high_first = self.bar_high_first(&bid_bar);
let high_leg = (
bid_bar.high,
ask_bar.high,
bid_sizes.high,
ask_sizes.high,
"bar high quote tick",
);
let low_leg = (
bid_bar.low,
ask_bar.low,
bid_sizes.low,
ask_sizes.low,
"bar low quote tick",
);
let legs = if high_first {
[high_leg, low_leg]
} else {
[low_leg, high_leg]
};
for (bid_price, ask_price, bid_size, ask_size, context) in legs {
self.fill_at_market = false;
quote_tick.bid_price = bid_price;
quote_tick.ask_price = ask_price;
quote_tick.bid_size = bid_size;
quote_tick.ask_size = ask_size;
if !self.process_bar_quote_tick(
"e_tick,
context,
&mut has_current_bid,
&mut has_current_ask,
) {
return;
}
}
self.fill_at_market = false;
quote_tick.bid_price = bid_bar.close;
quote_tick.ask_price = ask_bar.close;
quote_tick.bid_size = bid_sizes.close;
quote_tick.ask_size = ask_sizes.close;
if !self.process_bar_quote_tick(
"e_tick,
"bar close quote tick",
&mut has_current_bid,
&mut has_current_ask,
) {
return;
}
self.last_bar_bid = None;
self.last_bar_ask = None;
self.fill_at_market = true;
}
fn process_bar_quote_tick(
&mut self,
quote: &QuoteTick,
context: &str,
has_current_bid: &mut bool,
has_current_ask: &mut bool,
) -> bool {
let has_bid_size = quote.bid_size.non_zero();
let has_ask_size = quote.ask_size.non_zero();
let mut book_changed = false;
let mut bid_cleared = false;
let mut ask_cleared = false;
match (has_bid_size, has_ask_size) {
(true, true) => {
if !self.update_quote_tick_or_skip(quote, context) {
return false;
}
*has_current_bid = true;
*has_current_ask = true;
book_changed = true;
}
_ => {
if has_bid_size {
self.update_bar_quote_bid(quote);
*has_current_bid = true;
book_changed = true;
} else if !*has_current_bid {
self.clear_bar_quote_bid(quote);
*has_current_bid = true;
book_changed = true;
bid_cleared = true;
}
if has_ask_size {
self.update_bar_quote_ask(quote);
*has_current_ask = true;
book_changed = true;
} else if !*has_current_ask {
self.clear_bar_quote_ask(quote);
*has_current_ask = true;
book_changed = true;
ask_cleared = true;
}
}
}
if book_changed
&& let (Some(best_bid), Some(best_ask)) =
(self.book.best_bid_price(), self.book.best_ask_price())
&& best_bid > best_ask
{
if has_bid_size && !has_ask_size {
self.clear_bar_quote_ask(quote);
ask_cleared = true;
} else if has_ask_size && !has_bid_size {
self.clear_bar_quote_bid(quote);
bid_cleared = true;
}
}
if has_bid_size {
self.last_quote_bid = Some(quote.bid_price);
} else if bid_cleared {
self.last_quote_bid = None;
}
if has_ask_size {
self.last_quote_ask = Some(quote.ask_price);
} else if ask_cleared {
self.last_quote_ask = None;
}
if !book_changed {
return true;
}
self.reset_trade_liquidity(quote.ts_event);
self.iterate(quote.ts_init, AggressorSide::NoAggressor);
true
}
fn bar_high_first(&self, bar: &Bar) -> bool {
!self.config.bar_adaptive_high_low_ordering || bar.high - bar.open < bar.open - bar.low
}
fn update_bar_quote_bid(&mut self, quote: &QuoteTick) {
let bid = BookOrder::new(
OrderSide::Buy,
quote.bid_price,
quote.bid_size,
OrderSide::Buy as u64,
);
self.book
.add(bid, 0, self.book.sequence.saturating_add(1), quote.ts_event);
}
fn clear_bar_quote_bid(&mut self, quote: &QuoteTick) {
self.book
.clear_bids(self.book.sequence.saturating_add(1), quote.ts_event);
}
fn update_bar_quote_ask(&mut self, quote: &QuoteTick) {
let ask = BookOrder::new(
OrderSide::Sell,
quote.ask_price,
quote.ask_size,
OrderSide::Sell as u64,
);
self.book
.add(ask, 0, self.book.sequence.saturating_add(1), quote.ts_event);
}
fn clear_bar_quote_ask(&mut self, quote: &QuoteTick) {
self.book
.clear_asks(self.book.sequence.saturating_add(1), quote.ts_event);
}
pub fn process_trade_tick(&mut self, trade: &TradeTick) {
log::debug!("Processing {trade}");
if let Err(e) = self.check_price_precision(trade.price.precision, "trade price") {
self.log_precision_mismatch("trade tick", trade.instrument_id, &e);
return;
}
if let Err(e) = self.check_size_precision(trade.size.precision, "trade size") {
self.log_precision_mismatch("trade tick", trade.instrument_id, &e);
return;
}
self.precision_mismatch_streak = 0;
let price_raw = trade.price.raw();
if self.book_type == BookType::L1_MBP {
if trade.ts_event < self.book.ts_last {
log::warn!(
"Skipping stale trade: ts_event {} < book.ts_last {} for {}",
trade.ts_event,
self.book.ts_last,
self.book.instrument_id,
);
self.iterate(trade.ts_init, AggressorSide::NoAggressor);
return;
}
if !self.update_trade_tick_or_skip(trade, "trade tick") {
return;
}
}
self.core.set_last_raw(trade.price);
if !self.config.trade_execution {
if self.book_type == BookType::L1_MBP {
if let Some(bid) = self.book.best_bid_price() {
self.core.set_bid_raw(bid);
}
if let Some(ask) = self.book.best_ask_price() {
self.core.set_ask_raw(ask);
}
} else {
self.iterate_with_mode(
trade.ts_init,
AggressorSide::NoAggressor,
OrderMatchMode::StopTriggers(TriggerType::LastPrice),
);
}
return;
}
let aggressor_side = trade.aggressor_side;
match aggressor_side {
AggressorSide::Buy => {
if self.core.ask.is_none_or(|ask| trade.price > ask) {
self.core.set_ask_raw(trade.price);
}
if self.core.bid.is_none() {
self.core.set_bid_raw(trade.price);
}
}
AggressorSide::Sell => {
if self.core.bid.is_none_or(|bid| trade.price < bid) {
self.core.set_bid_raw(trade.price);
}
if self.core.ask.is_none() {
self.core.set_ask_raw(trade.price);
}
}
AggressorSide::NoAggressor => {
if self.core.bid.is_none_or(|bid| trade.price <= bid) {
self.core.set_bid_raw(trade.price);
}
if self.core.ask.is_none_or(|ask| trade.price >= ask) {
self.core.set_ask_raw(trade.price);
}
}
}
let original_bid = self.core.bid;
let original_ask = self.core.ask;
match aggressor_side {
AggressorSide::Sell => {
if original_ask.is_some_and(|ask| trade.price < ask) {
self.core.set_ask_raw(trade.price);
}
}
AggressorSide::Buy => {
if original_bid.is_some_and(|bid| trade.price > bid) {
self.core.set_bid_raw(trade.price);
}
}
AggressorSide::NoAggressor => {
self.core.set_bid_raw(trade.price);
self.core.set_ask_raw(trade.price);
}
}
self.start_trade_iteration(trade.size);
if self.config.liquidity_consumption && self.book_type != BookType::L1_MBP {
self.seed_trade_consumption(
price_raw,
trade.size.raw(),
trade.ts_event,
aggressor_side,
);
}
self.resolve_pending_on_trade(price_raw);
self.decrement_queue_on_trade(price_raw, trade.size.raw(), aggressor_side);
self.iterate(trade.ts_init, aggressor_side);
self.finish_trade_iteration();
if self.book_type == BookType::L1_MBP {
match aggressor_side {
AggressorSide::Sell => {
if let Some(ask) = self.last_quote_ask {
self.core.ask = Some(ask);
}
}
AggressorSide::Buy => {
if let Some(bid) = self.last_quote_bid {
self.core.bid = Some(bid);
}
}
AggressorSide::NoAggressor => {}
}
} else {
match aggressor_side {
AggressorSide::Sell => {
if let Some(ask) = original_ask
&& trade.price < ask
{
self.core.ask = Some(ask);
}
}
AggressorSide::Buy => {
if let Some(bid) = original_bid
&& trade.price > bid
{
self.core.bid = Some(bid);
}
}
AggressorSide::NoAggressor => {}
}
}
}
fn start_trade_iteration(&mut self, size: Quantity) {
self.last_trade_size = Some(size);
self.trade_consumption = 0;
self.l1_trade_size = (self.book_type == BookType::L1_MBP
&& self.config.liquidity_consumption)
.then_some(size.raw());
}
fn finish_trade_iteration(&mut self) {
self.last_trade_size = None;
if self.l1_trade_size.is_none() {
self.trade_consumption = 0;
}
}
fn reset_trade_liquidity(&mut self, timestamp: UnixNanos) {
if timestamp >= self.book.ts_last {
self.l1_trade_size = None;
self.trade_consumption = 0;
}
}
fn update_quote_tick_or_skip(&mut self, quote: &QuoteTick, context: &str) -> bool {
if let Err(e) = self.book.update_quote_tick(quote) {
log::warn!(
"Skipping {context} for {}: update_quote_tick failed: {e}",
quote.instrument_id,
);
return false;
}
self.reset_trade_liquidity(quote.ts_event);
true
}
fn update_trade_tick_or_skip(&mut self, trade: &TradeTick, context: &str) -> bool {
if let Err(e) = self.book.update_trade_tick(trade) {
log::warn!(
"Skipping {context} for {}: update_trade_tick failed: {e}",
trade.instrument_id,
);
return false;
}
self.reset_trade_liquidity(trade.ts_event);
true
}
pub fn process_status(&mut self, action: MarketStatusAction) {
log::debug!("Processing {action}");
match action {
MarketStatusAction::Trading | MarketStatusAction::PreOpen
if matches!(
self.market_status,
MarketStatus::Closed | MarketStatus::Paused | MarketStatus::Suspended
) =>
{
self.market_status = MarketStatus::Open;
}
MarketStatusAction::Pause if self.market_status == MarketStatus::Open => {
self.market_status = MarketStatus::Paused;
}
MarketStatusAction::Suspend if self.market_status == MarketStatus::Open => {
self.market_status = MarketStatus::Suspended;
}
MarketStatusAction::Halt | MarketStatusAction::Close
if self.market_status == MarketStatus::Open =>
{
self.market_status = MarketStatus::Closed;
}
_ => {}
}
}
pub fn process_instrument_close(&mut self, close: InstrumentClose) {
if close.instrument_id != self.instrument.id() {
log::warn!(
"Received instrument close for unknown instrument_id: {}",
close.instrument_id
);
return;
}
if close.close_type == InstrumentCloseType::ContractExpired {
self.instrument_close = Some(close);
self.iterate(close.ts_init, AggressorSide::NoAggressor);
}
}
pub fn process_instrument_expiration(&mut self, timestamp_ns: UnixNanos) {
self.check_instrument_expiration(timestamp_ns, false, &[]);
}
#[must_use]
pub const fn is_expiration_processed(&self) -> bool {
self.expiration_processed
}
pub fn has_pending_fills(&mut self) -> bool {
self.purge_applied_fills();
!self.pending_fills.is_empty()
}
fn requires_pending_resolution(&self) -> bool {
matches!(self.instrument, InstrumentAny::BinaryOption(_))
}
fn cancel_open_orders_for_expiration(&mut self, excluded: &[ClientOrderId]) {
let instrument_id = self.instrument.id();
let expiration_order_ids: IndexSet<ClientOrderId> = {
let cache = self.cache.borrow();
let mut order_ids = IndexSet::new();
for order_info in self.get_open_orders() {
order_ids.insert(order_info.client_order_id);
}
for order in cache.orders(None, Some(&instrument_id), None, None, None) {
if order.is_open() || order.is_inflight() {
order_ids.insert(order.client_order_id());
}
}
order_ids.retain(|client_order_id| !excluded.contains(client_order_id));
order_ids
};
for client_order_id in expiration_order_ids {
if let Some(order) = self.order_snapshot(client_order_id)
&& !self.contingent_order_is_closed(&order)
{
self.cancel_order_excluding(&order, None, excluded);
}
}
}
fn enter_pending_resolution(&mut self, excluded: &[ClientOrderId]) {
if self.pending_resolution {
return;
}
self.pending_resolution = true;
self.market_status = MarketStatus::Closed;
self.cancel_open_orders_for_expiration(excluded);
log::info!(
"{} expired and is now pending resolution; open orders canceled and new orders blocked",
self.instrument.id()
);
}
fn check_instrument_expiration(
&mut self,
timestamp_ns: UnixNanos,
defer_settlement: bool,
excluded: &[ClientOrderId],
) {
if self.expiration_processed || self.option_settlement_failed {
return;
}
let timestamp_triggered = self
.instrument
.expiration_ns()
.is_some_and(|ns| timestamp_ns >= ns);
if !timestamp_triggered && self.instrument_close.is_none() {
return;
}
if self.instrument_close.is_none()
&& timestamp_triggered
&& self.requires_pending_resolution()
{
self.enter_pending_resolution(excluded);
return;
}
self.enter_pending_resolution(excluded);
if self.has_pending_fills() {
return;
}
if matches!(
self.instrument,
InstrumentAny::OptionContract(_) | InstrumentAny::CryptoOption(_)
) {
if defer_settlement
&& self.instrument_close.is_none()
&& self.instrument.expiration_ns() == Some(timestamp_ns)
{
return;
}
match self.process_option_expiry(timestamp_ns) {
Ok(true) => {
self.expiration_processed = true;
self.pending_resolution = false;
self.instrument_close.take();
self.option_settlement_warning = None;
log::info!("{} reached expiration", self.instrument.id());
}
Ok(false) => {}
Err(e) => {
self.option_settlement_failed = true;
log::error!(
"Option settlement failed terminally for {}: {e}",
self.instrument.id()
);
}
}
return;
}
self.expiration_processed = true;
self.pending_resolution = false;
let close = self.instrument_close.take();
log::info!("{} reached expiration", self.instrument.id());
let instrument_id = self.instrument.id();
let positions: Vec<(
TraderId,
StrategyId,
AccountId,
PositionId,
OrderSide,
Quantity,
)> = {
let cache = self.cache.borrow();
cache
.positions_open(None, Some(&instrument_id), None, None, None)
.into_iter()
.filter_map(|pos| {
OrderCore::closing_side(pos.side).map(|closing_side| {
(
pos.trader_id,
pos.strategy_id,
pos.account_id,
pos.id,
closing_side,
pos.quantity,
)
})
})
.collect()
};
let ts_now = self.clock.borrow().timestamp_ns();
let close_price = close.as_ref().map(|close| close.close_price);
for (trader_id, strategy_id, account_id, position_id, closing_side, quantity) in positions {
let client_order_id =
ClientOrderId::from(format!("EXPIRATION-{}-{}", self.venue, UUID4::new()).as_str());
let mut order = OrderAny::Market(MarketOrder::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
closing_side,
quantity,
TimeInForce::Gtc,
UUID4::new(),
ts_now,
true, false,
None,
None,
None,
None,
None,
None,
None,
Some(vec![Ustr::from(&format!(
"EXPIRATION_{}_CLOSE",
self.venue
))]),
));
order.set_liquidity_side(LiquiditySide::Taker);
let add_result =
self.cache
.borrow_mut()
.add_order(order.clone(), Some(position_id), None, false);
if add_result.is_err() {
log::debug!("Expiration order already in cache: {client_order_id}");
} else {
self.publish_order_initialized(&order);
}
let venue_order_id = self.ids_generator.get_venue_order_id(&order).unwrap();
self.account_ids.insert(trader_id, account_id);
self.generate_order_accepted(&order, venue_order_id);
if let Some(fill_price) = close_price {
if let Err(e) = self.apply_fills(
&order,
&[(fill_price, quantity)],
LiquiditySide::Taker,
Some(position_id),
None,
None,
false,
) {
log::error!("Cannot fill expiration order {client_order_id}: {e}");
}
} else {
self.fill_market_order(client_order_id);
}
}
}
pub fn liquidate_open_positions(
&mut self,
ts_now: UnixNanos,
cancel_open_orders: bool,
settlement_currency: Currency,
) {
if self.instrument.settlement_currency() != settlement_currency {
return;
}
if cancel_open_orders {
let open_orders: Vec<RestingOrder> = self.get_open_orders();
for order_info in &open_orders {
let order = {
let cache = self.cache.borrow();
cache.order_owned(&order_info.client_order_id)
};
if let Some(order) = order {
self.cancel_order(&order, None);
}
}
}
let instrument_id = self.instrument.id();
let positions: Vec<(
TraderId,
StrategyId,
AccountId,
PositionId,
OrderSide,
Quantity,
)> = {
let cache = self.cache.borrow();
cache
.positions_open(None, Some(&instrument_id), None, None, None)
.into_iter()
.filter_map(|pos| {
OrderCore::closing_side(pos.side).map(|closing_side| {
(
pos.trader_id,
pos.strategy_id,
pos.account_id,
pos.id,
closing_side,
pos.quantity,
)
})
})
.collect()
};
for (trader_id, strategy_id, account_id, position_id, closing_side, quantity) in positions {
let has_price = if closing_side == OrderSide::Sell {
self.best_bid_price().is_some()
} else {
self.best_ask_price().is_some()
};
if !has_price {
log::warn!(
"LIQUIDATION: no price available for {instrument_id} position {position_id}, skipping"
);
continue;
}
let client_order_id = ClientOrderId::from(
format!("LIQUIDATION-{}-{}", self.venue, UUID4::new()).as_str(),
);
let order = OrderAny::Market(MarketOrder::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
closing_side,
quantity,
TimeInForce::Ioc,
UUID4::new(),
ts_now,
true, false,
None,
None,
None,
None,
None,
None,
None,
Some(vec![Ustr::from(&format!(
"LIQUIDATION_{}_CLOSE",
self.venue
))]),
));
let venue_order_id = self.ids_generator.get_venue_order_id(&order).unwrap();
{
let mut cache = self.cache.borrow_mut();
if let Err(e) = cache.add_order(order.clone(), Some(position_id), None, false) {
log::debug!("Liquidation order already in cache: {e}");
} else {
drop(cache);
self.publish_order_initialized(&order);
self.cache
.borrow_mut()
.add_venue_order_id(&client_order_id, &venue_order_id, false)
.ok();
}
}
self.account_ids.insert(trader_id, account_id);
self.generate_order_submitted(&order, account_id);
self.generate_order_accepted(&order, venue_order_id);
self.fill_market_order(client_order_id);
}
}
pub fn process_order(&mut self, order: &mut OrderAny, account_id: AccountId) {
self.process_order_received(order, account_id, false);
}
pub fn process_oto_command(&mut self, command: &OtoCommand) {
match command {
OtoCommand::Release {
client_order_id,
account_id,
..
} => {
if self
.orders
.closed_oto_order_ids
.borrow()
.contains(client_order_id)
|| self
.cache
.borrow()
.venue_order_id(client_order_id)
.is_some()
{
return;
}
let Some(mut order) = self.order_snapshot(*client_order_id) else {
return;
};
if let Err(e) = self.apply_deferred_submission(&mut order, false) {
log::error!("Cannot release OTO order {client_order_id}: {e}");
return;
}
if self.contingent_order_is_closed(&order)
|| self.inflight_orders.contains(*client_order_id)
{
return;
}
self.process_order_received(&mut order, *account_id, true);
}
OtoCommand::Resize {
parent_order_id,
quantity,
..
} => self.process_oto_resize(*parent_order_id, *quantity),
OtoCommand::Cancel {
client_order_id,
cancel_contingencies,
excluded,
..
} => {
let Some(mut order) = self.order_snapshot(*client_order_id) else {
return;
};
if self.contingent_order_is_closed(&order) {
return;
}
if let Err(e) = self.apply_deferred_submission(&mut order, false) {
log::error!("Cannot cancel OTO order {client_order_id}: {e}");
return;
}
self.cancel_order_excluding(&order, Some(*cancel_contingencies), excluded);
}
}
}
pub fn process_oto_resize(&mut self, parent_order_id: ClientOrderId, quantity: Quantity) {
if let Some(parent) = self.order_snapshot(parent_order_id) {
self.update_oto_children(&parent, Some(quantity), &[]);
}
}
fn process_order_received(
&mut self,
order: &mut OrderAny,
account_id: AccountId,
released: bool,
) {
if self.core.order_exists(order.client_order_id()) {
return;
}
if !self.cache.borrow().order_exists(&order.client_order_id()) {
log::error!(
"Cannot process order {}: not found in cache",
order.client_order_id()
);
return;
}
let ts_now = self.clock.borrow().timestamp_ns();
self.check_instrument_expiration(
ts_now,
self.config.defer_option_settlement,
&[order.client_order_id()],
);
if self.config.support_contingent_orders
&& order.parent_order_id().is_some()
&& !self.prepare_oto_submission(order, account_id)
{
return;
}
self.process_order_internal(order, account_id, released);
}
fn prepare_oto_submission(&mut self, order: &mut OrderAny, account_id: AccountId) -> bool {
let id = order.client_order_id();
let Some(snapshot) = self.order_snapshot(id) else {
return false;
};
if self.contingent_order_is_closed(&snapshot) {
return false;
}
*order = snapshot;
self.account_ids.insert(order.trader_id(), account_id);
self.orders.pending_oto_order_ids.borrow_mut().insert(id);
if let Err(e) = self.apply_deferred_submission(order, false) {
log::error!("Cannot process OTO submission {id}: {e}");
return false;
}
if let Some(command) = self.oto_submission_cancel(order) {
self.process_oto_command(&command);
return false;
}
self.resize_oto_order(order)
}
fn resize_oto_order(&mut self, order: &mut OrderAny) -> bool {
let Some(parent) = self.full_trigger_oto_parent(order) else {
return true;
};
if let Err(e) = self.apply_deferred_submission(order, false) {
log::error!("Cannot resize OTO order {}: {e}", order.client_order_id());
return false;
}
let ids = parent
.linked_order_ids()
.expect("OTO parent has linked children");
let filled_qty = self.engine_filled_qty(&parent);
let remaining = self.oto_child_remaining(order, ids, parent.quantity());
if remaining.is_zero() {
self.cancel_order(order, Some(false));
return false;
}
let parent_inverse = self
.cache
.borrow()
.instrument(&parent.instrument_id())
.is_some_and(Instrument::is_inverse);
let sizing = OtoSizing {
quantity: parent.quantity(),
filled_qty,
quote: parent.is_quote_quantity() && !parent_inverse,
reduced_to_filled: true,
grew: false,
};
self.resize_oto_child(order, &sizing, remaining);
if let Some(snapshot) = self.order_snapshot(order.client_order_id()) {
order.set_quantity(snapshot.quantity());
order.set_leaves_qty(snapshot.leaves_qty());
order.set_is_quote_quantity(snapshot.is_quote_quantity());
}
true
}
fn full_trigger_oto_parent(&self, order: &OrderAny) -> Option<OrderAny> {
if !self.config.support_contingent_orders || !self.config.oto_full_trigger {
return None;
}
let parent = self.order_snapshot(order.parent_order_id()?)?;
if parent.contingency_type() != Some(ContingencyType::Oto) {
return None;
}
let ids = parent.linked_order_ids()?;
if self.engine_filled_qty(&parent) < parent.quantity()
|| !ids
.iter()
.filter_map(|id| self.order_snapshot(*id))
.any(|child| child.instrument_id() != parent.instrument_id())
{
return None;
}
Some(parent)
}
fn process_order_internal(
&mut self,
order: &mut OrderAny,
account_id: AccountId,
released: bool,
) {
let reject_reason: Option<Ustr> = 'validate: {
let cache_borrow = self.cache.as_ref().borrow();
self.account_ids.insert(order.trader_id(), account_id);
if self.pending_resolution {
break 'validate Some(
format!(
"Contract {} has expired and is pending resolution",
self.instrument.id()
)
.into(),
);
}
if self.instrument.has_expiration() {
if let Some(activation_ns) = self.instrument.activation_ns()
&& self.clock.borrow().timestamp_ns() < activation_ns
{
break 'validate Some(
format!(
"Contract {} is not yet active, activation {activation_ns}",
self.instrument.id(),
)
.into(),
);
}
if let Some(expiration_ns) = self.instrument.expiration_ns()
&& self.clock.borrow().timestamp_ns() >= expiration_ns
{
break 'validate Some(
format!(
"Contract {} has expired, expiration {expiration_ns}",
self.instrument.id(),
)
.into(),
);
}
}
if self.expiration_processed {
break 'validate Some(
format!("Contract {} has expired", self.instrument.id()).into(),
);
}
if self.market_status != MarketStatus::Open {
break 'validate Some(
format!(
"Market {} is {}, cannot accept order {}",
self.instrument.id(),
self.market_status,
order.client_order_id()
)
.into(),
);
}
if self.config.support_contingent_orders {
if let Some(parent_order_id) = order.parent_order_id() {
self.orders
.pending_oto_order_ids
.borrow_mut()
.swap_remove(&order.client_order_id());
let parent_order = match self.order_snapshot(parent_order_id) {
Some(o) if o.contingency_type() == Some(ContingencyType::Oto) => o,
_ => panic!("OTO parent not found"),
};
let parent_filled_qty = parent_order.filled_qty();
if parent_order.status() == OrderStatus::Rejected
|| self
.orders
.rejected_oto_parent_ids
.borrow()
.contains(&parent_order_id)
{
break 'validate Some(
format!("Rejected OTO order from {parent_order_id}").into(),
);
} else if !released
&& (parent_filled_qty.is_zero()
|| (self.config.oto_full_trigger
&& parent_filled_qty < parent_order.quantity()))
{
log::info!(
"Pending OTO order {} triggers from {parent_order_id}",
order.client_order_id(),
);
self.orders
.pending_oto_order_ids
.borrow_mut()
.insert(order.client_order_id());
return;
}
}
if let Some(linked_order_ids) = order.linked_order_ids() {
let contingency_type = order.contingency_type();
for client_order_id in linked_order_ids {
match cache_borrow.order(client_order_id) {
Some(contingent_order)
if matches!(
contingency_type,
Some(ContingencyType::Oco | ContingencyType::Ouo)
) && !order.is_closed()
&& contingent_order.is_closed()
&& !Self::is_rejected_oto_sibling(order, &contingent_order) =>
{
break 'validate Some(
format!("Contingent order {client_order_id} already closed")
.into(),
);
}
None => panic!("Cannot find contingent order for {client_order_id}"),
_ => {}
}
}
}
}
if !order_precision_valid(order.quantity().precision, self.instrument.size_precision())
{
break 'validate Some(
format!(
"Invalid order quantity precision for order {}, was {} when {} size precision is {}",
order.client_order_id(),
order.quantity().precision,
self.instrument.id(),
self.instrument.size_precision()
)
.into(),
);
}
if let Some(display_qty) = order.display_qty()
&& !order_precision_valid(display_qty.precision, self.instrument.size_precision())
{
break 'validate Some(
format!(
"Invalid order display quantity precision for order {}, was {} when {} size precision is {}",
order.client_order_id(),
display_qty.precision,
self.instrument.id(),
self.instrument.size_precision()
)
.into(),
);
}
if let Some(reason) = order
.price()
.and_then(|price| self.order_price_rejection(order, "price", price))
{
break 'validate Some(reason);
}
if let Some(reason) = order
.trigger_price()
.and_then(|price| self.order_price_rejection(order, "trigger price", price))
{
break 'validate Some(reason);
}
if order.is_reduce_only() && !self.config.use_reduce_only {
break 'validate Some(
"Reduce-only orders are not supported by this matching engine".into(),
);
}
None
};
if let Some(reason) = reject_reason {
self.generate_order_rejected(order, reason);
return;
}
if let Some(reason) = self
.order_position_rejection(order)
.unwrap_or_else(|e| Some(e.to_string().into()))
{
self.generate_order_rejected(order, reason);
return;
}
if order.is_quote_quantity()
&& !self.instrument.is_inverse()
&& !matches!(
order.order_type(),
OrderType::TrailingStopLimit | OrderType::TrailingStopMarket,
)
&& (order.price().is_some()
|| matches!(
order.order_type(),
OrderType::Market | OrderType::MarketToLimit,
))
&& !self.convert_quote_to_base_quantity(order)
{
return;
}
match order.order_type() {
OrderType::Market => self.process_market_order(order),
OrderType::Limit => self.process_limit_order(order),
OrderType::MarketToLimit => self.process_market_to_limit_order(order),
OrderType::StopMarket => self.process_stop_market_order(order),
OrderType::StopLimit => self.process_stop_limit_order(order),
OrderType::MarketIfTouched => self.process_market_if_touched_order(order),
OrderType::LimitIfTouched => self.process_limit_if_touched_order(order),
OrderType::TrailingStopMarket => self.process_trailing_stop_order(order),
OrderType::TrailingStopLimit => self.process_trailing_stop_order(order),
}
}
fn is_rejected_oto_sibling(order: &OrderAny, sibling: &OrderAny) -> bool {
sibling.status() == OrderStatus::Rejected
&& order
.parent_order_id()
.is_some_and(|parent_id| sibling.parent_order_id() == Some(parent_id))
}
fn order_position_rejection(&mut self, order: &OrderAny) -> anyhow::Result<Option<Ustr>> {
let checks_reduce_only = self.config.use_reduce_only
&& order.is_reduce_only()
&& !order.is_closed()
&& !(self.restricts_short_selling() && order.is_sell());
if self.restricts_short_selling() || checks_reduce_only {
self.purge_applied_fills();
}
let cache = self.cache.borrow();
let position = self.position_for_order_in_cache(&cache, order);
if let Some(reason) = self.cash_sell_rejection(&cache, order, position.as_ref())? {
return Ok(Some(reason));
}
if checks_reduce_only {
let quantity = self.position_quantity(&cache, order, position.as_ref())?;
if (order.is_buy() && quantity >= Decimal::ZERO)
|| (order.is_sell() && quantity <= Decimal::ZERO)
{
return Ok(Some(
format!(
"Reduce-only order {} ({}-{}) would have increased position",
order.client_order_id(),
order.order_type().to_string().to_uppercase(),
order.order_side().to_string().to_uppercase()
)
.into(),
));
}
}
Ok(None)
}
fn restricts_short_selling(&self) -> bool {
self.account_type != AccountType::Margin
&& matches!(
self.instrument,
InstrumentAny::Equity(_) | InstrumentAny::BinaryOption(_)
)
}
fn cash_sell_rejection(
&self,
cache: &Cache,
order: &OrderAny,
position: Option<&Position>,
) -> anyhow::Result<Option<Ustr>> {
if !self.restricts_short_selling() || !order.is_sell() {
return Ok(None);
}
let quantity = if order.is_quote_quantity() {
let reference_price = order.price().or(self.core.bid).ok_or_else(|| {
anyhow::anyhow!(
"No market for {} to convert quote quantity to base",
order.instrument_id()
)
})?;
self.instrument
.try_calculate_base_quantity(order.leaves_qty(), reference_price)?
} else {
order.leaves_qty()
};
let position_quantity = self.position_quantity(cache, order, position)?;
if quantity.as_decimal() <= position_quantity {
return Ok(None);
}
let position_string = position.map_or("None".to_string(), |pos| pos.id.to_string());
Ok(Some(
format!(
"Short selling not permitted on a CASH account with position {position_string} and order {order}",
)
.into(),
))
}
fn order_price_rejection(&self, order: &OrderAny, field: &str, price: Price) -> Option<Ustr> {
let price_precision = self.instrument.price_precision();
if !order_precision_valid(price.precision, price_precision) {
return Some(
format!(
"Invalid order {field} precision for order {}, was {} when {} price precision is {price_precision}",
order.client_order_id(),
price.precision,
self.instrument.id(),
)
.into(),
);
}
let price_increment = self.instrument.price_increment();
if !Self::price_matches_tick(price, price_increment) {
return Some(
format!(
"Invalid order {field} {price} for order {}, not a multiple of {} price increment {price_increment}",
order.client_order_id(),
self.instrument.id(),
)
.into(),
);
}
None
}
fn convert_quote_to_base_quantity(&self, order: &mut OrderAny) -> bool {
let reference_price = if let Some(price) = order.price() {
Some(price)
} else {
match order.order_side() {
OrderSide::Buy => self.core.ask,
OrderSide::Sell => self.core.bid,
}
};
let Some(reference_price) = reference_price else {
self.generate_order_rejected(
order,
format!(
"No market for {} to convert quote quantity to base",
order.instrument_id(),
)
.into(),
);
return false;
};
let base_quantity = self
.instrument
.calculate_base_quantity(order.quantity(), reference_price);
let ts_now = self.clock.borrow().timestamp_ns();
let update = OrderUpdated::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
base_quantity,
UUID4::new(),
ts_now,
ts_now,
false,
order.venue_order_id(),
order.account_id(),
None,
None,
None,
false,
);
let event = OrderEventAny::Updated(update);
if let Err(e) = order.apply(event.clone()) {
log::error!(
"Failed to apply quote-to-base update for {}: {e}",
order.client_order_id(),
);
return false;
}
self.orders
.pending_updates
.borrow_mut()
.entry(order.client_order_id())
.or_default()
.push(update);
self.dispatch_order_event(event);
true
}
pub fn restore_open_order(&mut self, order: &OrderAny, account_id: AccountId) {
if self.core.order_exists(order.client_order_id()) {
return;
}
if self.contingent_order_is_closed(order) {
return;
}
self.account_ids.insert(order.trader_id(), account_id);
if let Err(e) = self.ids_generator.get_venue_order_id(order) {
log::error!(
"Failed to resolve venue order ID restoring {}: {e}",
order.client_order_id(),
);
return;
}
if let Some(mut cached) = self.cache.borrow_mut().order_mut(&order.client_order_id())
&& !matches!(
cached.liquidity_side(),
Some(LiquiditySide::Maker | LiquiditySide::Taker)
)
{
cached.set_liquidity_side(LiquiditySide::Maker);
}
let filled_qty = order.filled_qty();
if !filled_qty.is_zero() {
self.orders
.cached_filled_qty
.borrow_mut()
.insert(order.client_order_id(), filled_qty);
}
let match_info = Self::matching_core_entry(order);
self.track_post_match_order(order);
self.core.add_order(match_info);
}
pub fn process_modify(&mut self, command: &ModifyOrder, account_id: AccountId) {
if !self.core.order_exists(command.client_order_id) {
self.generate_order_modify_rejected(
command.trader_id,
command.strategy_id,
command.instrument_id,
command.client_order_id,
Ustr::from(format!("Order {} not found", command.client_order_id).as_str()),
command.venue_order_id,
Some(account_id),
);
return;
}
let order = match self.order_snapshot(command.client_order_id) {
Some(order) => order,
None => {
log::error!(
"Cannot modify order: order {} not found in cache",
command.client_order_id
);
return;
}
};
let update_success = self.update_order(
&order,
command.quantity,
command.price,
command.trigger_price,
None,
);
if !update_success {
return;
}
if !self.core.order_exists(command.client_order_id) {
return;
}
let Some(refreshed) = self.resync_core_entry(command.client_order_id) else {
return;
};
let price_changed = refreshed.price() != order.price()
|| refreshed.trigger_price() != order.trigger_price();
if price_changed
&& refreshed.is_open()
&& self.config.queue_position
&& let Some(new_price) = refreshed.price()
{
self.snapshot_queue_position(&refreshed, new_price);
self.queue_excess.swap_remove(&refreshed.client_order_id());
}
}
pub fn process_cancel(&mut self, command: &CancelOrder, account_id: AccountId) {
if !self.core.order_exists(command.client_order_id)
&& !self
.orders
.pending_oto_order_ids
.borrow()
.contains(&command.client_order_id)
{
let order = self
.order_snapshot(command.client_order_id)
.filter(|order| {
order.account_id() == Some(account_id)
&& (order.is_inflight() || order.is_open())
&& !self.inflight_orders.contains(order.client_order_id())
});
if let Some(order) = order {
self.cancel_order(&order, None);
return;
}
self.generate_order_cancel_rejected(
command.trader_id,
command.strategy_id,
account_id,
command.instrument_id,
command.client_order_id,
command.venue_order_id,
Ustr::from(format!("Order {} not found", command.client_order_id).as_str()),
);
return;
}
let mut order = match self.order_snapshot(command.client_order_id) {
Some(order) => order,
None => {
log::error!(
"Cannot cancel order: order {} not found in cache",
command.client_order_id
);
return;
}
};
if let Err(e) = self.apply_deferred_submission(&mut order, false) {
self.generate_order_cancel_rejected(
command.trader_id,
command.strategy_id,
account_id,
command.instrument_id,
command.client_order_id,
command.venue_order_id,
Ustr::from(e.to_string().as_str()),
);
return;
}
if !order.is_inflight() && !order.is_open() {
self.purge_stale_core_entry(command.client_order_id);
return;
}
self.cancel_order(&order, None);
}
pub fn process_cancel_all(&mut self, command: &CancelAllOrders, account_id: AccountId) {
self.process_cancel_all_excluding(command, account_id, &[]);
}
pub fn process_cancel_all_excluding(
&mut self,
command: &CancelAllOrders,
account_id: AccountId,
excluded: &[ClientOrderId],
) {
let instrument_id = command.instrument_id;
let order_side = command.order_side;
let mut client_order_ids: Vec<ClientOrderId> = {
let cache = self.cache.borrow();
cache
.orders_open_refs(
None,
Some(&instrument_id),
None,
Some(&account_id),
order_side,
)
.into_iter()
.chain(cache.orders_inflight_refs(
None,
Some(&instrument_id),
None,
Some(&account_id),
order_side,
))
.map(|order| order.client_order_id())
.filter(|client_order_id| !excluded.contains(client_order_id))
.collect()
};
client_order_ids.sort_unstable();
client_order_ids.dedup();
for client_order_id in client_order_ids {
let order = match self
.cache
.borrow()
.order(&client_order_id)
.map(|o| o.clone())
{
Some(order) => order,
None => continue,
};
if !order.is_inflight() && !order.is_open() {
self.purge_stale_core_entry(client_order_id);
continue;
}
self.cancel_order_excluding(&order, None, excluded);
}
}
fn purge_stale_core_entry(&mut self, client_order_id: ClientOrderId) {
if self.core.order_exists(client_order_id) {
self.delete_core_order(client_order_id);
}
self.remove_queue_position(client_order_id);
self.orders
.cached_filled_qty
.borrow_mut()
.swap_remove(&client_order_id);
}
fn resync_core_entry(&mut self, client_order_id: ClientOrderId) -> Option<OrderAny> {
let order = self.order_snapshot(client_order_id)?;
if order.is_closed() {
self.delete_core_order(client_order_id);
self.remove_queue_position(client_order_id);
return Some(order);
}
let new_match_info = Self::matching_core_entry(&order);
let unchanged = self
.core
.get_order(client_order_id)
.is_some_and(|existing| *existing == new_match_info);
if unchanged {
self.track_post_match_order(&order);
return Some(order);
}
self.delete_core_order(client_order_id);
self.track_post_match_order(&order);
self.core.add_order(new_match_info);
Some(order)
}
fn order_snapshot(&self, client_order_id: ClientOrderId) -> Option<OrderAny> {
let mut order = self.cache.borrow().order(&client_order_id)?.clone();
let mut pending = self.orders.pending_updates.borrow_mut();
if order.is_closed() {
pending.swap_remove(&client_order_id);
return Some(order);
}
if let Some(updates) = pending.get_mut(&client_order_id) {
Self::retain_unapplied_updates(&order, updates);
for update in updates.iter() {
if let Err(e) = order.apply(OrderEventAny::Updated(*update)) {
log::error!("Cannot apply pending update for {client_order_id}: {e}");
return None;
}
}
if updates.is_empty() {
pending.swap_remove(&client_order_id);
}
}
if let Some(filled_qty) = self
.orders
.cached_filled_qty
.borrow()
.get(&client_order_id)
.copied()
{
write_filled_qty(&mut order, filled_qty);
order.set_leaves_qty(order.quantity().saturating_sub(filled_qty));
}
Some(order)
}
fn purge_applied_updates(&self) {
let cache = self.cache.borrow();
self.orders
.pending_updates
.borrow_mut()
.retain(|id, updates| {
let Some(order) = cache.order(id) else {
return false;
};
Self::retain_unapplied_updates(&order, updates);
!updates.is_empty()
});
}
fn purge_applied_rejections(&self) {
let cache = self.cache.borrow();
self.orders
.rejected_oto_parent_ids
.borrow_mut()
.retain(|id| {
cache
.order(id)
.is_some_and(|order| order.status() != OrderStatus::Rejected)
});
}
fn purge_applied_oto_terminals(&self) {
let cache = self.cache.borrow();
self.orders
.closed_oto_order_ids
.borrow_mut()
.retain(|id| cache.order(id).is_some_and(|order| !order.is_closed()));
}
fn retain_unapplied_updates(order: &OrderAny, updates: &mut Vec<OrderUpdated>) {
if order.is_closed() {
updates.clear();
return;
}
let events = order.events();
updates.retain(|update| {
!events.iter().any(|event| {
matches!(event, OrderEventAny::Updated(applied) if applied.event_id == update.event_id)
})
});
}
pub fn process_batch_cancel(&mut self, command: &BatchCancelOrders, account_id: AccountId) {
for order in &command.cancels {
self.process_cancel(order, account_id);
}
}
pub fn process_batch_modify(&mut self, command: &BatchModifyOrders, account_id: AccountId) {
for order in &command.modifies {
self.process_modify(order, account_id);
}
}
fn process_market_order(&mut self, order: &OrderAny) {
if order.time_in_force() == TimeInForce::AtTheOpen
|| order.time_in_force() == TimeInForce::AtTheClose
{
self.generate_order_rejected(
order,
format!(
"time in force {} is not currently supported",
order.time_in_force()
)
.into(),
);
return;
}
if (order.order_side() == OrderSide::Buy && self.core.ask.is_none())
|| (order.order_side() == OrderSide::Sell && self.core.bid.is_none())
{
self.generate_order_rejected(
order,
format!("No market for {}", order.instrument_id()).into(),
);
return;
}
if self.config.use_market_order_acks {
let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
self.generate_order_accepted(order, venue_order_id);
}
self.fill_market_order(order.client_order_id());
}
fn process_limit_order(&mut self, order: &mut OrderAny) {
if order.time_in_force() == TimeInForce::AtTheOpen
|| order.time_in_force() == TimeInForce::AtTheClose
{
self.generate_order_rejected(
order,
format!(
"time in force {} is not currently supported",
order.time_in_force()
)
.into(),
);
return;
}
let limit_px = order.price().expect("Limit order must have a price");
if order.is_post_only() && self.core.is_limit_matched(order.order_side(), limit_px) {
self.generate_order_rejected(
order,
format!(
"POST_ONLY {} {} order limit px of {} would have been a TAKER: bid={}, ask={}",
order.order_type(),
order.order_side(),
order.price().unwrap(),
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.into(),
);
return;
}
self.accept_order(order);
if self.core.is_limit_matched(order.order_side(), limit_px) {
order.set_liquidity_side(LiquiditySide::Taker);
self.fill_limit_order_with_snapshot(order.clone());
if self.core.order_exists(order.client_order_id())
&& let Some(mut order) = self.cache.borrow_mut().order_mut(&order.client_order_id())
{
order.set_liquidity_side(LiquiditySide::Maker);
}
} else if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc) {
self.cancel_order(order, None);
} else {
if let Some(price) = order.price() {
self.snapshot_queue_position(order, price);
}
if let Some(mut order) = self.cache.borrow_mut().order_mut(&order.client_order_id())
&& !matches!(
order.liquidity_side(),
Some(LiquiditySide::Maker | LiquiditySide::Taker)
)
{
order.set_liquidity_side(LiquiditySide::Maker);
}
}
}
fn process_market_to_limit_order(&mut self, order: &OrderAny) {
if (order.order_side() == OrderSide::Buy && self.core.ask.is_none())
|| (order.order_side() == OrderSide::Sell && self.core.bid.is_none())
{
self.generate_order_rejected(
order,
format!("No market for {}", order.instrument_id()).into(),
);
return;
}
if self.config.use_market_order_acks {
let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
self.generate_order_accepted(order, venue_order_id);
}
let client_order_id = order.client_order_id();
self.fill_market_order(client_order_id);
let filled_qty = self
.orders
.cached_filled_qty
.borrow()
.get(&client_order_id)
.copied()
.unwrap_or_default();
let leaves_qty = order.quantity().saturating_sub(filled_qty);
if leaves_qty.is_zero()
|| order.time_in_force() == TimeInForce::Fok
|| (order.time_in_force() == TimeInForce::Ioc && filled_qty.is_zero())
{
self.purge_cached_filled_qty_if_closed(client_order_id);
return;
}
if let Some(mut updated_order) = self.order_snapshot(client_order_id) {
self.accept_order(&mut updated_order);
}
}
fn process_stop_market_order(&mut self, order: &mut OrderAny) {
let stop_px = order
.trigger_price()
.expect("Stop order must have a trigger price");
if self.core.is_stop_matched_with_trigger_type(
order.order_side(),
stop_px,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
if self.config.reject_stop_orders {
self.generate_order_rejected(order, self.in_market_rejection(order, "stop"));
return;
}
self.fill_market_order(order.client_order_id());
return;
}
self.accept_order(order);
}
fn process_stop_limit_order(&mut self, order: &mut OrderAny) {
let stop_px = order
.trigger_price()
.expect("Stop order must have a trigger price");
if self.core.is_stop_matched_with_trigger_type(
order.order_side(),
stop_px,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
if self.config.reject_stop_orders {
self.generate_order_rejected(order, self.in_market_rejection(order, "stop"));
return;
}
self.accept_triggered_limit_style_order(order);
return;
}
self.accept_order(order);
}
fn process_market_if_touched_order(&mut self, order: &mut OrderAny) {
if self.core.is_touch_triggered_with_trigger_type(
order.order_side(),
order.trigger_price().unwrap(),
order.trigger_type().unwrap_or(TriggerType::Default),
) {
if self.config.reject_stop_orders {
self.generate_order_rejected(order, self.in_market_rejection(order, "trigger"));
return;
}
self.fill_market_order(order.client_order_id());
return;
}
self.accept_order(order);
}
fn process_limit_if_touched_order(&mut self, order: &mut OrderAny) {
if self.core.is_touch_triggered_with_trigger_type(
order.order_side(),
order.trigger_price().unwrap(),
order.trigger_type().unwrap_or(TriggerType::Default),
) {
if self.config.reject_stop_orders {
self.generate_order_rejected(order, self.in_market_rejection(order, "trigger"));
return;
}
self.accept_triggered_limit_style_order(order);
return;
}
self.accept_order(order);
}
fn in_market_rejection(&self, order: &OrderAny, field: &str) -> Ustr {
format!(
"{} {} order {field} px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
order.order_type(),
order.order_side(),
order.trigger_price().expect("Triggered order must have a trigger price"),
self.core.bid.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core.ask.map_or_else(|| "None".to_string(), |p| p.to_string()),
)
.into()
}
fn accept_triggered_limit_style_order(&mut self, order: &mut OrderAny) {
self.accept_order(order);
self.trigger_limit_style_stop_order(order.client_order_id(), order.clone());
if let Some(cached_order) = self
.cache
.borrow()
.order(&order.client_order_id())
.map(|order| order.clone())
{
*order = cached_order;
}
}
fn process_trailing_stop_order(&mut self, order: &mut OrderAny) {
let side = order.order_side();
let trigger_type = order.trigger_type().unwrap_or(TriggerType::Default);
if let Some(activation_price) = order.activation_price()
&& self.trailing_stop_activates_now(order)
&& self.config.reject_stop_orders
{
self.generate_order_rejected(
order,
format!(
"{} {} order activation px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
order.order_type(),
side,
activation_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.into(),
);
return;
}
if self.trailing_stop_activates_now(order)
&& let Some(trigger_price) = order.trigger_price()
&& self
.core
.is_stop_matched_with_trigger_type(side, trigger_price, trigger_type)
{
self.generate_order_rejected(
order,
format!(
"{} {} order trigger px of {} was in the market: bid={}, ask={}, but rejected because of configuration",
order.order_type(),
side,
trigger_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.into(),
);
return;
}
order.set_liquidity_side(LiquiditySide::Maker);
self.accept_order(order);
}
fn trailing_stop_activates_now(&self, order: &OrderAny) -> bool {
let side = order.order_side();
match order.activation_price() {
Some(activation_price) => self.core.is_touch_triggered_with_trigger_type(
side,
activation_price,
if order.trigger_type() == Some(TriggerType::MarkPrice) {
TriggerType::MarkPrice
} else {
TriggerType::BidAsk
},
),
None => self
.get_trailing_activation_price(
order.trigger_type().unwrap_or(TriggerType::Default),
side,
self.core.bid,
self.core.ask,
self.core.last,
)
.is_some(),
}
}
pub fn iterate(&mut self, timestamp_ns: UnixNanos, aggressor_side: AggressorSide) {
self.iterate_with_mode(timestamp_ns, aggressor_side, OrderMatchMode::All);
}
fn iterate_with_mode(
&mut self,
timestamp_ns: UnixNanos,
aggressor_side: AggressorSide,
match_mode: OrderMatchMode,
) {
self.purge_closed_cached_filled_qty();
self.purge_applied_updates();
self.purge_applied_fills();
self.purge_applied_rejections();
self.purge_applied_oto_terminals();
if aggressor_side == AggressorSide::NoAggressor && self.last_trade_size.is_none() {
if self.book_type == BookType::L1_MBP {
if let Some(bid) = self.book.best_bid_price() {
self.core.set_bid_raw(bid);
}
if let Some(ask) = self.book.best_ask_price() {
self.core.set_ask_raw(ask);
}
} else {
self.core.bid = self.book.best_bid_price();
self.core.ask = self.book.best_ask_price();
}
}
let mut matched_order = false;
if self.market_status == MarketStatus::Open {
for action in self.core.iterate_bids() {
if !self.should_process_match_action(action, match_mode) {
continue;
}
matched_order = true;
match action {
MatchAction::FillLimit(id) => self.fill_resting_limit_order(id),
MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
}
}
for action in self.core.iterate_asks() {
if !self.should_process_match_action(action, match_mode) {
continue;
}
matched_order = true;
match action {
MatchAction::FillLimit(id) => self.fill_resting_limit_order(id),
MatchAction::TriggerStop(id) => self.trigger_stop_order(id),
}
}
}
let order_ids: Vec<ClientOrderId> = if matched_order {
self.core.iter_orders().map(|m| m.client_order_id).collect()
} else if self.post_match_order_ids.is_empty() {
Vec::new()
} else {
self.core
.iter_orders()
.filter_map(|order| {
self.post_match_order_ids
.contains(&order.client_order_id)
.then_some(order.client_order_id)
})
.collect()
};
let support_gtd_orders = self.config.support_gtd_orders;
for client_order_id in order_ids {
let (action, keep_tracking) = {
let cache = self.cache.borrow();
let Some(order) = cache.order(&client_order_id) else {
self.post_match_order_ids.swap_remove(&client_order_id);
continue;
};
(
post_match_order_action(&order, support_gtd_orders, timestamp_ns, |order| {
self.order_snapshot(client_order_id)
.unwrap_or_else(|| order.clone())
}),
Self::requires_post_match_maintenance(&order),
)
};
match action {
PostMatchOrderAction::RemoveClosed => {
self.delete_core_order(client_order_id);
self.remove_queue_position(client_order_id);
self.orders
.cached_filled_qty
.borrow_mut()
.swap_remove(&client_order_id);
continue;
}
PostMatchOrderAction::Expire(order) => {
self.delete_core_order(client_order_id);
self.expire_order(&order);
self.purge_cached_filled_qty_if_closed(client_order_id);
continue;
}
PostMatchOrderAction::UpdateTrailing(mut order) => {
if self.maybe_activate_trailing_stop(
&mut order,
self.core.bid,
self.core.ask,
self.core.last,
) {
self.update_trailing_stop_order(&order);
self.resync_core_entry(client_order_id);
}
}
PostMatchOrderAction::NoMaintenance => {
if !keep_tracking {
self.post_match_order_ids.swap_remove(&client_order_id);
}
}
}
if self.target_bid.is_some() || self.target_ask.is_some() || self.target_last.is_some()
{
if let Some(t) = self.target_bid.take() {
self.core.bid = Some(t);
}
if let Some(t) = self.target_ask.take() {
self.core.ask = Some(t);
}
if let Some(t) = self.target_last.take() {
self.core.last = Some(t);
}
}
}
if let Some(t) = self.target_bid.take() {
self.core.bid = Some(t);
}
if let Some(t) = self.target_ask.take() {
self.core.ask = Some(t);
}
if let Some(t) = self.target_last.take() {
self.core.last = Some(t);
}
self.core.bid = self.book.best_bid_price();
self.core.ask = self.book.best_ask_price();
self.check_instrument_expiration(timestamp_ns, self.config.defer_option_settlement, &[]);
self.purge_closed_cached_filled_qty();
self.purge_applied_updates();
self.purge_applied_fills();
self.purge_applied_rejections();
self.purge_applied_oto_terminals();
}
fn fill_resting_limit_order(&mut self, client_order_id: ClientOrderId) {
if self
.core
.get_order(client_order_id)
.is_some_and(|order| order.order_type == OrderType::MarketToLimit)
&& let Some(mut order) = self.cache.borrow_mut().order_mut(&client_order_id)
{
order.set_liquidity_side(LiquiditySide::Maker);
}
self.fill_limit_order(client_order_id);
}
fn should_process_match_action(&self, action: MatchAction, match_mode: OrderMatchMode) -> bool {
let client_order_id = match action {
MatchAction::FillLimit(id) | MatchAction::TriggerStop(id) => id,
};
if !self.core.order_exists(client_order_id) {
return false;
}
match match_mode {
OrderMatchMode::All => true,
OrderMatchMode::StopTriggers(trigger_type) => match action {
MatchAction::TriggerStop(client_order_id) => self
.core
.get_order(client_order_id)
.is_some_and(|order| order.trigger_type == Some(trigger_type)),
MatchAction::FillLimit(_) => false,
},
}
}
fn get_trailing_activation_price(
&self,
trigger_type: TriggerType,
order_side: OrderSide,
bid: Option<Price>,
ask: Option<Price>,
last: Option<Price>,
) -> Option<Price> {
match trigger_type {
TriggerType::LastPrice => last,
TriggerType::MarkPrice => self.core.mark,
TriggerType::LastOrBidAsk => last.or(match order_side {
OrderSide::Buy => ask,
OrderSide::Sell => bid,
}),
_ => match order_side {
OrderSide::Buy => ask,
OrderSide::Sell => bid,
},
}
}
fn maybe_activate_trailing_stop(
&self,
order: &mut OrderAny,
bid: Option<Price>,
ask: Option<Price>,
last: Option<Price>,
) -> bool {
let is_activated = match order {
OrderAny::TrailingStopMarket(inner) => inner.is_activated,
OrderAny::TrailingStopLimit(inner) => inner.is_activated,
_ => return true,
};
if is_activated {
return true;
}
let (bid, ask) = if order.trigger_type() == Some(TriggerType::MarkPrice) {
(self.core.mark, self.core.mark)
} else {
(bid, ask)
};
let activation_price = match order.activation_price() {
Some(price) => {
let hit = match order.order_side() {
OrderSide::Buy => ask.is_some_and(|ask| ask <= price),
OrderSide::Sell => bid.is_some_and(|bid| bid >= price),
};
hit.then_some(price)
}
None => self.get_trailing_activation_price(
order
.trigger_type()
.expect("Trailing order has a trigger type"),
order.order_side(),
bid,
ask,
last,
),
};
let Some(activation_price) = activation_price else {
return false;
};
match order {
OrderAny::TrailingStopMarket(inner) => {
inner.activation_price = Some(activation_price);
inner.set_activated();
}
OrderAny::TrailingStopLimit(inner) => {
inner.activation_price = Some(activation_price);
inner.set_activated();
}
_ => unreachable!("only trailing orders require activation"),
}
store_trailing_stop_activation(&mut self.cache.borrow_mut(), order);
true
}
fn determine_limit_price_and_volume(
&mut self,
order: &OrderAny,
leaves_qty: Quantity,
journal: Option<&mut ConsumptionJournal>,
) -> Vec<(Price, Quantity)> {
match order.price() {
Some(order_price) => {
let mut fills = if self.config.liquidity_consumption {
let size_prec = self.instrument.size_precision();
self.book
.get_all_crossed_levels(order.order_side(), order_price, size_prec)
} else {
let book_order =
BookOrder::new(order.order_side(), order_price, order.quantity(), 1);
self.book.simulate_fills(&book_order)
};
if let Some(trade_size) = self.last_trade_size
&& let Some(trade_price) = self.core.last
{
let fills_at_trade_price = fills.iter().any(|(px, _)| *px == trade_price);
if (self.book_type == BookType::L1_MBP || !fills_at_trade_price)
&& self.core.is_limit_matched(order.order_side(), order_price)
{
let available_qty = if self.config.liquidity_consumption {
let remaining = trade_size.raw().saturating_sub(self.trade_consumption);
Quantity::from_raw(remaining, trade_size.precision)
} else {
trade_size
};
let fill_qty = min(leaves_qty, available_qty);
if fill_qty.non_zero() {
let fill_price = if self.book_type == BookType::L1_MBP
&& fills_at_trade_price
&& order.liquidity_side() == Some(LiquiditySide::Taker)
{
trade_price
} else {
order_price
};
log::debug!(
"Trade execution fill: {} @ {} (trade_price={}, available: {}, book had {} fills)",
fill_qty,
fill_price,
trade_price,
available_qty,
fills.len()
);
if self.config.liquidity_consumption {
self.trade_consumption += fill_qty.raw();
}
return vec![(fill_price, fill_qty)];
}
if self.book_type == BookType::L1_MBP {
return Vec::new();
}
}
}
if fills.is_empty() {
return fills;
}
let book_prices: Vec<Price> = if self.config.liquidity_consumption {
fills.iter().map(|(px, _)| *px).collect()
} else {
Vec::new()
};
let book_prices_ref: Option<&[Price]> = if book_prices.is_empty() {
None
} else {
Some(&book_prices)
};
if order
.liquidity_side()
.is_some_and(|liquidity_side| liquidity_side == LiquiditySide::Maker)
{
let trigger_price = order
.trigger_price()
.filter(|_| order.trigger_type() != Some(TriggerType::MarkPrice));
match order.order_side() {
OrderSide::Buy => {
let target_price = if trigger_price
.is_some_and(|trigger_price| order_price > trigger_price)
{
trigger_price.unwrap()
} else {
order_price
};
for fill in &mut fills {
let last_px = fill.0;
if last_px < order_price {
self.target_bid = self.core.bid;
self.target_ask = self.core.ask;
self.target_last = self.core.last;
self.core.set_ask_raw(target_price);
self.core.set_last_raw(target_price);
fill.0 = target_price;
}
}
}
OrderSide::Sell => {
let target_price = if trigger_price
.is_some_and(|trigger_price| order_price < trigger_price)
{
trigger_price.unwrap()
} else {
order_price
};
for fill in &mut fills {
let last_px = fill.0;
if last_px > order_price {
self.target_bid = self.core.bid;
self.target_ask = self.core.ask;
self.target_last = self.core.last;
self.core.set_bid_raw(target_price);
self.core.set_last_raw(target_price);
fill.0 = target_price;
}
}
}
}
}
self.apply_liquidity_consumption(
fills,
order,
leaves_qty,
book_prices_ref,
self.l1_trade_size,
journal,
)
}
None => panic!("Limit order must have a price"),
}
}
fn determine_market_price_and_volume(&self, order: &OrderAny) -> Vec<(Price, Quantity)> {
let price = match order.order_side() {
OrderSide::Buy => Price::max(FIXED_PRECISION),
OrderSide::Sell => Price::min(FIXED_PRECISION),
};
let mut fills = if self.config.liquidity_consumption {
let size_prec = self.instrument.size_precision();
self.book
.get_all_crossed_levels(order.order_side(), price, size_prec)
} else {
let book_order = BookOrder::new(order.order_side(), price, order.quantity(), 0);
self.book.simulate_fills(&book_order)
};
if !self.fill_at_market
&& order.trigger_type() != Some(TriggerType::MarkPrice)
&& self.book_type == BookType::L1_MBP
&& !fills.is_empty()
&& matches!(
order.order_type(),
OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched
)
&& let Some(trigger_price) = order.trigger_price()
{
fills[0] = (trigger_price, fills[0].1);
let mut remaining_qty = order.leaves_qty();
let mut capped_fills = Vec::with_capacity(fills.len());
for (price, qty) in fills {
if remaining_qty.is_zero() {
break;
}
let mut capped_qty = qty.min(remaining_qty);
capped_qty.precision = qty.precision;
if capped_qty.is_zero() {
continue;
}
remaining_qty = remaining_qty - capped_qty;
capped_fills.push((price, capped_qty));
}
return capped_fills;
}
fills
}
fn determine_market_fill_model_price_and_volume(
&mut self,
order: &OrderAny,
) -> anyhow::Result<(Vec<(Price, Quantity)>, bool)> {
if let Some(book) = self.fill_model.get_orderbook_for_fill_simulation(
&self.instrument,
order,
self.core.bid,
self.core.ask,
)? {
let price = match order.order_side() {
OrderSide::Buy => Price::max(FIXED_PRECISION),
OrderSide::Sell => Price::min(FIXED_PRECISION),
};
let book_order = BookOrder::new(order.order_side(), price, order.quantity(), 0);
return Ok((book.simulate_fills(&book_order), true));
}
Ok((self.determine_market_price_and_volume(order), false))
}
fn determine_limit_fill_model_price_and_volume(
&mut self,
order: &OrderAny,
leaves_qty: Quantity,
journal: Option<&mut ConsumptionJournal>,
) -> anyhow::Result<Vec<(Price, Quantity)>> {
if let Some(book) = self.fill_model.get_orderbook_for_fill_simulation(
&self.instrument,
order,
self.core.bid,
self.core.ask,
)? && let Some(limit_price) = order.price()
{
let book_order = BookOrder::new(order.order_side(), limit_price, order.quantity(), 0);
return Ok(book.simulate_fills(&book_order));
}
Ok(self.determine_limit_price_and_volume(order, leaves_qty, journal))
}
pub fn fill_market_order(&mut self, client_order_id: ClientOrderId) {
let mut order = match self.order_snapshot(client_order_id) {
Some(order) => order,
None => {
log::error!("Cannot fill market order: order {client_order_id} not found in cache");
return;
}
};
if !self.prepare_order_fill(&mut order) {
return;
}
if order.is_quote_quantity()
&& !self.instrument.is_inverse()
&& !self.convert_quote_to_base_quantity(&mut order)
{
return;
}
if let Some(filled_qty) = self
.orders
.cached_filled_qty
.borrow()
.get(&order.client_order_id())
.copied()
&& filled_qty >= order.quantity()
{
log::debug!(
"Ignoring fill as already filled pending application of events: {:?}, {:?}, {:?}, {:?}",
filled_qty,
order.quantity(),
order.filled_qty(),
order.quantity()
);
return;
}
let (venue_position_id, position) = self.fill_position_for_order(&order, Some(true));
let Some(leaves_qty) = self.fill_leaves_qty(&order, position.as_ref()) else {
return;
};
order.set_liquidity_side(LiquiditySide::Taker);
let (mut fills, from_synthetic) =
match self.determine_market_fill_model_price_and_volume(&order) {
Ok(result) => result,
Err(e) => {
log::error!(
"Cannot fill market order {}: fill model failed: {e}",
order.client_order_id()
);
return;
}
};
let protection_price: Option<Price> = if let Some(protection_points) =
self.config.price_protection_points
&& matches!(
order.order_type(),
OrderType::Market | OrderType::StopMarket
) {
protection_price_calculate(
self.instrument.price_increment(),
&order,
protection_points,
self.core.bid,
self.core.ask,
)
.ok()
} else {
None
};
if let Some(protection_price) = protection_price {
fills = self.filter_fills_by_protection(fills, &order, protection_price);
}
let is_trigger_price_fill = !self.fill_at_market
&& order.trigger_type() != Some(TriggerType::MarkPrice)
&& self.book_type == BookType::L1_MBP
&& matches!(
order.order_type(),
OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched
)
&& order.trigger_price().is_some();
if !from_synthetic && !is_trigger_price_fill {
let mut journal = self.fok_consumption_journal(&order);
fills = self.apply_liquidity_consumption(
fills,
&order,
leaves_qty,
None,
None,
journal.as_mut(),
);
if let Some(journal) = journal
&& self.is_fok_unfillable(&order, &fills, leaves_qty)
{
self.revert_consumption_journal(journal);
}
}
if let Err(e) = self.apply_fills(
&order,
&fills,
LiquiditySide::Taker,
if self.config.use_reduce_only && order.is_reduce_only() {
venue_position_id
} else {
None
},
position.as_ref(),
protection_price,
from_synthetic,
) {
log::error!("Cannot fill market order {}: {e}", order.client_order_id());
}
}
fn filter_fills_by_protection(
&self,
fills: Vec<(Price, Quantity)>,
order: &OrderAny,
protection_price: Price,
) -> Vec<(Price, Quantity)> {
fills
.into_iter()
.filter(|(fill_price, _)| {
match order.order_side() {
OrderSide::Buy => *fill_price <= protection_price,
OrderSide::Sell => *fill_price >= protection_price,
}
})
.collect()
}
pub fn fill_limit_order(&mut self, client_order_id: ClientOrderId) {
let order = match self.order_snapshot(client_order_id) {
Some(order) => order,
None => {
log::error!("Cannot fill limit order: order {client_order_id} not found in cache");
return;
}
};
self.fill_limit_order_with_snapshot(order);
}
fn fill_limit_order_with_snapshot(&mut self, mut order: OrderAny) {
if !self.prepare_order_fill(&mut order) {
return;
}
if order.is_quote_quantity()
&& !self.instrument.is_inverse()
&& !self.convert_quote_to_base_quantity(&mut order)
{
return;
}
match order.price() {
Some(order_price) => {
let cached_filled_qty = self
.orders
.cached_filled_qty
.borrow()
.get(&order.client_order_id())
.copied();
if let Some(qty) = cached_filled_qty
&& qty >= order.quantity()
{
log::debug!(
"Ignoring fill as already filled pending application of events: {}, {}, {}, {}",
qty,
order.quantity(),
order.filled_qty(),
order.leaves_qty(),
);
return;
}
if order
.liquidity_side()
.is_some_and(|liquidity_side| liquidity_side == LiquiditySide::Maker)
{
let at_limit = if self.last_trade_size.is_some() && self.core.last.is_some() {
self.core.last.is_some_and(|last| last == order_price)
} else if order.order_side() == OrderSide::Buy {
self.core.bid.is_some_and(|bid| bid == order_price)
} else {
self.core.ask.is_some_and(|ask| ask == order_price)
};
if at_limit {
let is_limit_filled = match self.fill_model.is_limit_filled() {
Ok(value) => value,
Err(e) => {
log::error!(
"Cannot fill limit order {}: fill model failed: {e}",
order.client_order_id()
);
return;
}
};
if !is_limit_filled {
return; }
}
}
let queue_allowed_raw = if self.config.queue_position {
match self.determine_trade_fill_qty(&order) {
None | Some(0) => {
if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc)
{
self.cancel_order(&order, None);
}
return;
}
Some(allowed) => Some(allowed),
}
} else {
None
};
let (venue_position_id, position) = self.fill_position_for_order(&order, None);
let Some(leaves_qty) = self.fill_leaves_qty(&order, position.as_ref()) else {
return;
};
let required_qty = leaves_qty;
let leaves_qty = queue_allowed_raw.map_or(leaves_qty, |allowed| {
Quantity::from_raw(leaves_qty.raw().min(allowed), leaves_qty.precision)
});
let mut journal = self.fok_consumption_journal(&order);
let tc_before = self.trade_consumption;
let mut fills = match self.determine_limit_fill_model_price_and_volume(
&order,
leaves_qty,
journal.as_mut(),
) {
Ok(fills) => fills,
Err(e) => {
log::error!(
"Cannot fill limit order {}: fill model failed: {e}",
order.client_order_id()
);
return;
}
};
if let Some(allowed_raw) = queue_allowed_raw {
let size_prec = self.instrument.size_precision();
let mut remaining = allowed_raw;
fills = fills
.into_iter()
.filter_map(|(price, qty)| {
if remaining == 0 {
return None;
}
let capped = qty.raw().min(remaining);
remaining -= capped;
Some((price, Quantity::from_raw(capped, size_prec)))
})
.collect();
let consumed: QuantityRaw = fills.iter().map(|(_, qty)| qty.raw()).sum();
if let Some(excess) = self.queue_excess.get_mut(&order.client_order_id()) {
*excess = excess.saturating_sub(consumed);
}
self.trade_consumption = tc_before + consumed;
}
if let Some(journal) = journal
&& self.is_fok_unfillable(&order, &fills, required_qty)
{
self.revert_consumption_journal(journal);
}
if fills.is_empty() {
log::debug!(
"Skipping fill for {}: no liquidity available",
order.client_order_id()
);
if matches!(order.time_in_force(), TimeInForce::Fok | TimeInForce::Ioc) {
self.cancel_order(&order, None);
}
return;
}
let liquidity_side = order.liquidity_side().unwrap();
if let Err(e) = self.apply_fills(
&order,
&fills,
liquidity_side,
venue_position_id,
position.as_ref(),
None,
false,
) {
log::error!("Cannot fill limit order {}: {e}", order.client_order_id());
}
}
None => panic!("Limit order must have a price"),
}
}
fn fill_leaves_qty(
&mut self,
order: &OrderAny,
position: Option<&Position>,
) -> Option<Quantity> {
if !self.config.liquidity_consumption && position.is_some() {
return Some(order.leaves_qty());
}
let leaves_qty = order
.quantity()
.saturating_sub(self.engine_filled_qty(order));
match self.reduce_only_remaining(order, position) {
Ok(Some(remaining)) if remaining.is_zero() => {
log::warn!(
"Canceling REDUCE_ONLY {} as would increase position",
order.order_type()
);
self.cancel_order(order, None);
None
}
Ok(Some(remaining)) => Some(leaves_qty.min(remaining)),
Ok(None) => Some(leaves_qty),
Err(e) => {
log::error!(
"Cannot determine fill quantity for {}: {e}",
order.client_order_id()
);
None
}
}
}
fn prepare_order_fill(&mut self, order: &mut OrderAny) -> bool {
if order.is_closed() {
self.purge_stale_core_entry(order.client_order_id());
return false;
}
self.resize_oto_order(order)
}
fn fill_position_for_order(
&mut self,
order: &OrderAny,
generate: Option<bool>,
) -> (Option<PositionId>, Option<Position>) {
if self.restricts_short_selling()
&& self.oms_type == OmsType::Hedging
&& let Some(position) = self.position_for_order_in_cache(&self.cache.borrow(), order)
{
return (Some(position.id), Some(position));
}
if self.restricts_short_selling()
&& self.oms_type == OmsType::Hedging
&& order.parent_order_id().is_some()
{
let cache = self.cache.borrow();
return (Self::position_id_for_order(&cache, order), None);
}
if self.oms_type == OmsType::Hedging
&& self.config.use_reduce_only
&& order.is_reduce_only()
{
let cache = self.cache.as_ref().borrow();
if let Some(position_id) = Self::position_id_for_order(&cache, order) {
let position = cache
.position(&position_id)
.map(|position| position.clone_without_events());
return (Some(position_id), position);
}
if self.config.support_contingent_orders && order.parent_order_id().is_some() {
return (None, None);
}
if let Some(position) = Self::open_position_reduced_by_order(&cache, order) {
return (Some(position.id), Some(position));
}
}
let pending_position_id = if self.oms_type == OmsType::Hedging {
self.pending_fills.values().find_map(|fill| {
(fill.client_order_id == order.client_order_id())
.then_some(fill.position_id)
.flatten()
})
} else {
None
};
let venue_position_id =
pending_position_id.or_else(|| self.ids_generator.get_position_id(order, generate));
let position = {
let cache = self.cache.as_ref().borrow();
venue_position_id
.as_ref()
.and_then(|position_id| cache.position(position_id))
.map(|position| position.clone_without_events())
};
(venue_position_id, position)
}
fn position_for_order_in_cache(&self, cache: &Cache, order: &OrderAny) -> Option<Position> {
if let Some(position_id) = Self::position_id_for_order(cache, order) {
return cache
.position(&position_id)
.map(|position| position.clone_without_events());
}
if self.oms_type == OmsType::Netting {
let position_id = PositionId::new(
format!("{}-{}", order.instrument_id(), order.strategy_id()).as_str(),
);
return cache
.position(&position_id)
.map(|position| position.clone_without_events());
}
if self.oms_type == OmsType::Hedging
&& self.config.use_reduce_only
&& order.is_reduce_only()
{
if self.config.support_contingent_orders && order.parent_order_id().is_some() {
return None;
}
return Self::open_position_reduced_by_order(cache, order);
}
None
}
fn position_id_for_order(cache: &Cache, order: &OrderAny) -> Option<PositionId> {
cache
.position_id(&order.client_order_id())
.copied()
.or_else(|| {
order
.parent_order_id()
.filter(|parent_id| {
cache
.order(parent_id)
.is_some_and(|parent| parent.instrument_id() == order.instrument_id())
})
.and_then(|parent_id| cache.position_id(&parent_id).copied())
})
}
fn open_position_reduced_by_order(cache: &Cache, order: &OrderAny) -> Option<Position> {
cache
.positions_open(
None,
Some(&order.instrument_id()),
Some(&order.strategy_id()),
None,
None,
)
.into_iter()
.find(|position| order.would_reduce_only(position.side, position.quantity))
.map(|position| position.clone_without_events())
}
#[allow(
clippy::too_many_arguments,
reason = "fill application needs order, position, pricing, and liquidity-source context"
)]
fn apply_fills(
&mut self,
order: &OrderAny,
fills: &[(Price, Quantity)],
liquidity_side: LiquiditySide,
venue_position_id: Option<PositionId>,
position: Option<&Position>,
protection_price: Option<Price>,
from_synthetic: bool,
) -> anyhow::Result<()> {
let mut reduce_only_remaining = self.reduce_only_remaining(order, position)?;
let required_qty = reduce_only_remaining.map_or(order.leaves_qty(), |remaining| {
order.leaves_qty().min(remaining)
});
if self.is_fok_unfillable(order, fills, required_qty) {
self.cancel_order(order, None);
return Ok(());
}
if fills.is_empty() {
match (order.status(), order.time_in_force()) {
(OrderStatus::Initialized | OrderStatus::Submitted, time_in_force)
if time_in_force != TimeInForce::Ioc =>
{
self.generate_order_rejected(
order,
format!("No market for {}", order.instrument_id()).into(),
);
}
_ => self.cancel_order(order, None),
}
return Ok(());
}
let venue_position_id = if self.oms_type == OmsType::Netting {
None
} else {
venue_position_id
};
let initial_market_to_limit =
order.order_type() == OrderType::MarketToLimit && order.filled_qty().is_zero();
let mut first_price = None;
let mut total_filled = self.engine_filled_qty(order);
let initial_total_filled = total_filled;
let mut last_fill_px: Option<Price> = None;
let mut reduce_only_filled = None;
if let Some(remaining) = reduce_only_remaining {
if remaining.is_zero() {
self.cancel_order(order, None);
return Ok(());
}
reduce_only_filled = Some(total_filled);
}
for &(fill_px, fill_qty) in fills {
let Some(mut fill_px) = self.normalize_fill_price(fill_px, order.client_order_id())
else {
continue;
};
let Some(fill_qty) = self.normalize_fill_quantity(fill_qty, order.client_order_id())
else {
continue;
};
if initial_market_to_limit {
if fill_qty.is_zero() {
continue;
}
if first_price.is_some_and(|price| price != fill_px) {
break;
}
if first_price.is_none() {
self.generate_order_updated(order, order.quantity(), Some(fill_px), None, None);
first_price = Some(fill_px);
}
}
if self.book_type == BookType::L1_MBP
&& self.fill_model.is_slipped()?
&& let Some(slipped_px) = self.slipped_price(fill_px, order.order_side())
{
fill_px = slipped_px;
}
let mut effective_fill_qty = fill_qty;
if let Some(remaining) = reduce_only_remaining {
if remaining.is_zero() {
return Ok(());
}
if effective_fill_qty > remaining {
let precision = effective_fill_qty.precision;
effective_fill_qty = remaining;
effective_fill_qty.precision = precision;
}
}
if fill_qty.is_zero() {
if fills.len() == 1 && order.status() == OrderStatus::Submitted {
self.generate_order_rejected(
order,
format!("No market for {}", order.instrument_id()).into(),
);
}
return Ok(());
}
let capped_fill_qty = min(
effective_fill_qty,
order.quantity().saturating_sub(total_filled),
);
let reduce_only_exhausts_position =
reduce_only_remaining.is_some_and(|remaining| capped_fill_qty >= remaining);
if reduce_only_exhausts_position {
let mut reduce_only_target = reduce_only_filled
.unwrap_or(initial_total_filled)
.checked_add(capped_fill_qty)
.expect("Overflow occurred when adding reduce-only target quantity");
reduce_only_target.precision = self.instrument.size_precision();
if order.quantity() != reduce_only_target {
self.generate_order_updated(order, reduce_only_target, None, None, None);
}
}
total_filled = total_filled.add(capped_fill_qty);
if let Some(remaining) = reduce_only_remaining.as_mut() {
*remaining = *remaining - capped_fill_qty.min(*remaining);
}
if let Some(filled) = reduce_only_filled.as_mut() {
*filled = filled
.checked_add(capped_fill_qty)
.expect("Overflow occurred when adding reduce-only filled quantity");
}
self.fill_order(
order,
fill_px,
effective_fill_qty,
liquidity_side,
venue_position_id,
position,
)?;
last_fill_px = Some(fill_px);
if initial_market_to_limit
&& (total_filled >= order.quantity() || reduce_only_exhausts_position)
{
return Ok(());
}
if reduce_only_exhausts_position {
self.purge_cached_filled_qty_if_closed(order.client_order_id());
return Ok(());
}
}
let filled_in_loop = total_filled > initial_total_filled;
if initial_market_to_limit && filled_in_loop {
return Ok(());
}
let leaves_remaining = total_filled < order.quantity();
if order.time_in_force() == TimeInForce::Ioc && leaves_remaining {
self.cancel_order(order, None);
return Ok(());
}
if leaves_remaining
&& !from_synthetic
&& (order.is_open() || filled_in_loop)
&& self.book_type == BookType::L1_MBP
&& matches!(
order.order_type(),
OrderType::Market
| OrderType::MarketIfTouched
| OrderType::StopMarket
| OrderType::TrailingStopMarket
)
{
let Some(last_fill_px) = last_fill_px else {
return Ok(());
};
let side = order.order_side();
let Some(slip_fill_px) = self.slipped_price(last_fill_px, side) else {
return Ok(());
};
if let Some(protection_price) = protection_price {
let exceeds_boundary = match side {
OrderSide::Buy => slip_fill_px > protection_price,
OrderSide::Sell => slip_fill_px < protection_price,
};
if exceeds_boundary {
return Ok(());
}
}
let mut leaves_qty = order.quantity().saturating_sub(total_filled);
if let Some(remaining) = reduce_only_remaining {
if remaining.is_zero() {
return Ok(());
}
if leaves_qty > remaining {
let precision = leaves_qty.precision;
leaves_qty = remaining;
leaves_qty.precision = precision;
}
if leaves_qty >= remaining {
let mut reduce_only_target = reduce_only_filled
.unwrap_or(initial_total_filled)
.checked_add(leaves_qty)
.expect("Overflow occurred when adding reduce-only target quantity");
reduce_only_target.precision = self.instrument.size_precision();
if order.quantity() != reduce_only_target {
self.generate_order_updated(order, reduce_only_target, None, None, None);
}
}
}
if leaves_qty.is_zero() {
return Ok(());
}
self.fill_order(
order,
slip_fill_px,
leaves_qty,
liquidity_side,
venue_position_id,
position,
)?;
self.purge_cached_filled_qty_if_closed(order.client_order_id());
}
Ok(())
}
fn slipped_price(&self, price: Price, side: OrderSide) -> Option<Price> {
let price_increment = self.instrument.price_increment();
let slipped = match side {
OrderSide::Buy => price.add(price_increment),
OrderSide::Sell => price.sub(price_increment),
};
is_price_tradable(&self.instrument, slipped).then_some(slipped)
}
fn normalize_fill_price(
&self,
fill_px: Price,
client_order_id: ClientOrderId,
) -> Option<Price> {
let normalized = self.normalize_price_for_current_instrument(fill_px);
if normalized.is_none() {
log::warn!(
"Skipping fill for {client_order_id}: fill price {fill_px} is not compatible \
with {} price_precision={} price_increment={}",
self.instrument.id(),
self.instrument.price_precision(),
self.instrument.price_increment()
);
}
normalized
}
fn normalize_fill_quantity(
&self,
fill_qty: Quantity,
client_order_id: ClientOrderId,
) -> Option<Quantity> {
let normalized = self.normalize_quantity_for_current_instrument(fill_qty);
if normalized.is_none() {
log::warn!(
"Skipping fill for {client_order_id}: fill quantity {fill_qty} is not compatible \
with {} size_precision={}",
self.instrument.id(),
self.instrument.size_precision()
);
}
normalized
}
fn reduce_only_remaining(
&mut self,
order: &OrderAny,
position: Option<&Position>,
) -> anyhow::Result<Option<Quantity>> {
if self.config.use_reduce_only && order.is_reduce_only() {
return Ok(Some(self.position_quantity_remaining(order, position)?));
}
Ok(None)
}
fn position_quantity_remaining(
&mut self,
order: &OrderAny,
position: Option<&Position>,
) -> anyhow::Result<Quantity> {
let precision = self.instrument.size_precision();
self.purge_applied_fills();
let quantity = self.position_quantity(&self.cache.borrow(), order, position)?;
if (order.is_buy() && quantity >= Decimal::ZERO)
|| (order.is_sell() && quantity <= Decimal::ZERO)
{
return Ok(Quantity::zero(precision));
}
Ok(Quantity::from_decimal_dp(quantity.abs(), precision)?)
}
fn position_quantity(
&self,
cache: &Cache,
order: &OrderAny,
position: Option<&Position>,
) -> anyhow::Result<Decimal> {
let position_id = position
.map(|position| position.id)
.or_else(|| Self::position_id_for_order(cache, order))
.or_else(|| {
(self.oms_type == OmsType::Netting).then(|| {
PositionId::new(format!("{}-{}", order.instrument_id(), order.strategy_id()))
})
});
let cached_position = position_id.as_ref().and_then(|id| cache.position(id));
let position = position.or(cached_position.as_deref());
let splits_virtual_position = self.restricts_short_selling()
&& self.oms_type == OmsType::Hedging
&& position_id.is_some_and(|id| id.is_virtual());
let mut quantity = position.map_or(Decimal::ZERO, |position| match position.side {
PositionSide::Long => position.quantity.as_decimal(),
PositionSide::Short => -position.quantity.as_decimal(),
PositionSide::Flat => Decimal::ZERO,
});
let mut flipped_orders = IndexSet::new();
for fill in self.pending_fills.values() {
let matches_position = match position_id {
Some(position_id) => fill.position_id == Some(position_id),
None => order.parent_order_id().is_some_and(|parent_id| {
fill.client_order_id == parent_id
|| cache
.order(&fill.client_order_id)
.is_some_and(|order| order.parent_order_id() == Some(parent_id))
}),
};
if !matches_position || flipped_orders.contains(&fill.client_order_id) {
continue;
}
let remaining = quantity
.checked_add(fill.quantity_change)
.ok_or_else(|| anyhow::anyhow!("Pending position quantity overflow"))?;
quantity = if splits_virtual_position
&& !quantity.is_zero()
&& !remaining.is_zero()
&& quantity.is_sign_negative() != remaining.is_sign_negative()
{
flipped_orders.insert(fill.client_order_id);
Decimal::ZERO
} else {
remaining
};
}
Ok(quantity)
}
fn purge_applied_fills(&mut self) {
let cache = self.cache.borrow();
let cash_cache = self.restricts_short_selling().then_some(&*cache);
self.pending_fills.retain(|trade_id, fill| {
let previous_position_id = fill.position_id;
fill.position_id = cache
.position_id(&fill.client_order_id)
.copied()
.or_else(|| {
fill.position_id
.filter(|id| cache.position(id).is_some())
})
.or_else(|| {
let cache = cash_cache?;
let order = cache.order(&fill.client_order_id)?;
Self::position_id_for_order(cache, &order)
})
.or(fill.position_id);
let Some(position_id) = fill.position_id else {
return cache.order_exists(&fill.client_order_id);
};
let Some(position) = cache.position(&position_id) else {
return cache.order_exists(&fill.client_order_id);
};
if previous_position_id
.and_then(|id| cache.position(&id))
.iter()
.chain(std::iter::once(&position))
.any(|position| position.trade_ids.contains(trade_id))
{
return false;
}
let opening_trade_id = position.events.first().map(|event| event.trade_id);
if opening_trade_id != fill.opening_trade_id {
if position.replay_events.iter().any(|event| {
matches!(event, PositionReplayEvent::Filled(event) if event.trade_id == *trade_id)
}) || cache.position_snapshots(Some(&position_id), None).iter()
.any(|snapshot| snapshot.trade_ids.contains(trade_id))
|| cash_cache
.and_then(|cache| cache.order(&fill.client_order_id))
.filter(|order| position.opening_order_id == order.client_order_id())
.is_some_and(|order| {
order
.trade_ids()
.into_iter()
.take_while(|id| Some(**id) != opening_trade_id)
.any(|id| id == trade_id)
})
{
return false;
}
fill.opening_trade_id = opening_trade_id;
}
true
});
}
fn fill_order(
&mut self,
order: &OrderAny,
last_px: Price,
last_qty: Quantity,
liquidity_side: LiquiditySide,
venue_position_id: Option<PositionId>,
position: Option<&Position>,
) -> anyhow::Result<()> {
self.check_size_precision(last_qty.precision, "fill quantity")?;
let (last_qty, new_filled_qty) = if let Some(filled_qty) = self
.orders
.cached_filled_qty
.borrow()
.get(&order.client_order_id())
.copied()
{
let leaves_qty = order.quantity().saturating_sub(filled_qty);
let last_qty = min(last_qty, leaves_qty);
(last_qty, filled_qty + last_qty)
} else {
let last_qty = min(last_qty, order.quantity());
(last_qty, last_qty)
};
if last_qty.is_zero() {
return Ok(());
}
let fee_order;
let commission_order = {
let mut cloned = order.clone();
write_filled_qty(&mut cloned, new_filled_qty.saturating_sub(last_qty));
if order.liquidity_side() != Some(liquidity_side) {
cloned.set_liquidity_side(liquidity_side);
}
fee_order = cloned;
&fee_order
};
let underlying_px = self.fee_underlying_price()?;
let commission = self.fee_model.get_commission_with_context(
commission_order,
last_qty,
last_px,
&self.instrument,
underlying_px,
)?;
let mut reduce_only_order_ids = position
.map(|position| self.reduce_only_order_ids(position.id))
.unwrap_or_default();
self.orders
.cached_filled_qty
.borrow_mut()
.insert(order.client_order_id(), new_filled_qty);
let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
self.generate_order_filled(
order,
venue_order_id,
venue_position_id,
last_qty,
last_px,
self.instrument.quote_currency(),
commission,
liquidity_side,
);
let post_fill_filled_qty = self.engine_filled_qty(order);
let post_fill_quantity = self
.orders
.pending_updates
.borrow()
.get(&order.client_order_id())
.and_then(|updates| updates.last())
.map_or_else(
|| {
self.cache
.borrow()
.order(&order.client_order_id())
.map_or(order.quantity(), |order| order.quantity())
},
|update| update.quantity,
);
let post_fill_leaves_qty = post_fill_quantity.saturating_sub(post_fill_filled_qty);
let fully_filled = post_fill_leaves_qty.is_zero();
if order.is_closed() || fully_filled {
self.delete_core_order(order.client_order_id());
self.remove_queue_position(order.client_order_id());
if order.order_type() != OrderType::MarketToLimit {
self.purge_cached_filled_qty_if_closed(order.client_order_id());
}
}
self.process_fill_contingencies(order, venue_position_id, post_fill_leaves_qty)?;
if let Some(position) = position {
reduce_only_order_ids.extend(self.reduce_only_order_ids(position.id));
}
reduce_only_order_ids.extend(self.reduce_only_linked_order_ids(
order,
venue_position_id.or(position.map(|position| position.id)),
));
reduce_only_order_ids.sort_unstable();
reduce_only_order_ids.dedup();
self.sync_reduce_only_orders(order, position, &reduce_only_order_ids)?;
Ok(())
}
fn process_fill_contingencies(
&mut self,
order: &OrderAny,
position_id: Option<PositionId>,
leaves_qty: Quantity,
) -> anyhow::Result<()> {
if !self.config.support_contingent_orders {
return Ok(());
}
self.queue_oto_resize(order);
match order.contingency_type() {
Some(ContingencyType::Oto) => {
self.release_oto_children_after_fill(order, position_id, leaves_qty.is_zero())?;
}
Some(ContingencyType::Oco) => self.cancel_oco_after_fill(order)?,
Some(ContingencyType::Ouo) => self.update_ouo_after_fill(order, leaves_qty)?,
None => {}
}
Ok(())
}
fn queue_oto_resize(&self, order: &OrderAny) {
let Some(parent) = self.full_trigger_oto_parent(order) else {
return;
};
let mut commands = self.orders.commands.borrow_mut();
let index = commands
.iter()
.position(|command| !matches!(command, OtoCommand::Cancel { .. }))
.unwrap_or(commands.len());
commands.insert(
index,
OtoCommand::Resize {
instrument_id: parent.instrument_id(),
parent_order_id: parent.client_order_id(),
quantity: parent.quantity(),
},
);
}
fn release_oto_children_after_fill(
&mut self,
parent: &OrderAny,
position_id: Option<PositionId>,
fully_filled: bool,
) -> anyhow::Result<()> {
let Some(ids) = parent.linked_order_ids() else {
log::error!(
"OTO order {} does not have linked orders",
parent.client_order_id()
);
return Ok(());
};
for id in ids {
let mut child = self
.order_snapshot(*id)
.ok_or_else(|| anyhow::anyhow!("Order {id} not found in cache"))?;
self.release_oto_child_after_fill(parent, &mut child, position_id, fully_filled)?;
}
Ok(())
}
fn release_oto_child_after_fill(
&mut self,
parent: &OrderAny,
child: &mut OrderAny,
position_id: Option<PositionId>,
fully_filled: bool,
) -> anyhow::Result<()> {
self.apply_deferred_submission(child, false)?;
let id = child.client_order_id();
if child.is_closed()
|| child.leaves_qty().is_zero()
|| child.is_active_local()
|| self.orders.closed_oto_order_ids.borrow().contains(&id)
|| self.inflight_orders.contains(id)
{
return Ok(());
}
if child.instrument_id() == parent.instrument_id()
&& let (None, Some(position_id)) =
(child.position_id(), parent.position_id().or(position_id))
{
self.cache
.borrow_mut()
.add_position_id(&position_id, &self.venue, &id, &child.strategy_id())
.unwrap();
log::debug!("Added position id {position_id} to cache for order {id}");
}
if child.is_open()
&& !(matches!(child.status(), OrderStatus::PendingUpdate)
&& child
.previous_status()
.is_some_and(|s| matches!(s, OrderStatus::Submitted)))
{
return Ok(());
}
let account_id = parent
.account_id()
.or_else(|| self.account_ids.get(&parent.trader_id()).copied())
.ok_or_else(|| {
anyhow::anyhow!("Account ID not found for trader {}", parent.trader_id())
})?;
if !self.config.oto_full_trigger || fully_filled {
self.release_oto_child(child, account_id, parent, None);
}
Ok(())
}
fn oto_submission_cancel(&self, order: &OrderAny) -> Option<OtoCommand> {
let id = order.client_order_id();
if let Some(command) = self.orders.commands.borrow().iter().find(|command| {
matches!(command, OtoCommand::Cancel { client_order_id, .. } if *client_order_id == id)
}) {
return Some(command.clone());
}
let sibling_filled = order.contingency_type() == Some(ContingencyType::Oco)
&& order.linked_order_ids().is_some_and(|ids| {
ids.iter().any(|id| {
self.order_snapshot(*id)
.is_some_and(|sibling| !self.engine_filled_qty(&sibling).is_zero())
})
});
sibling_filled.then(|| OtoCommand::Cancel {
instrument_id: order.instrument_id(),
client_order_id: id,
cancel_contingencies: false,
excluded: Vec::new(),
})
}
fn cancel_oco_after_fill(&mut self, order: &OrderAny) -> anyhow::Result<()> {
let Some(ids) = order.linked_order_ids() else {
log::error!(
"OCO order {} does not have linked orders",
order.client_order_id()
);
return Ok(());
};
for id in ids {
let child = self
.order_snapshot(*id)
.ok_or_else(|| anyhow::anyhow!("Order {id} not found in cache"))?;
if self.contingent_order_is_closed(&child) || child.is_active_local() {
continue;
}
self.cancel_order(&child, Some(false));
}
Ok(())
}
fn update_ouo_after_fill(
&mut self,
order: &OrderAny,
leaves_qty: Quantity,
) -> anyhow::Result<()> {
let Some(ids) = order.linked_order_ids() else {
log::error!(
"OUO order {} does not have linked orders",
order.client_order_id()
);
return Ok(());
};
let mut excluded = ids.to_vec();
excluded.push(order.client_order_id());
for id in ids {
let child = self
.order_snapshot(*id)
.ok_or_else(|| anyhow::anyhow!("Order {id} not found in cache"))?;
self.update_ouo_child_after_fill(&child, leaves_qty, &excluded);
}
Ok(())
}
fn update_ouo_child_after_fill(
&mut self,
child: &OrderAny,
leaves_qty: Quantity,
excluded: &[ClientOrderId],
) {
if child.is_active_local()
|| self.contingent_order_is_closed(child)
|| self.inflight_orders.contains(child.client_order_id())
{
return;
}
let leaves_qty = self.quantity_for_order(child, leaves_qty);
let child_filled_qty = self.engine_filled_qty(child);
if leaves_qty.is_zero() && !child.is_closed() {
self.cancel_order_excluding(child, None, excluded);
} else if child_filled_qty >= leaves_qty {
self.cancel_order(child, Some(false));
} else if leaves_qty.non_zero() {
self.update_contingent_quantity(child, leaves_qty, false);
}
}
fn reduce_only_linked_order_ids(
&self,
order: &OrderAny,
position_id: Option<PositionId>,
) -> Vec<ClientOrderId> {
if !self.config.support_contingent_orders || !self.config.use_reduce_only {
return Vec::new();
}
let Some(linked_order_ids) = order.linked_order_ids() else {
return Vec::new();
};
let cache = self.cache.borrow();
let position_id = position_id
.or_else(|| Self::position_id_for_order(&cache, order))
.or_else(|| {
self.position_for_order_in_cache(&cache, order)
.map(|position| position.id)
});
linked_order_ids
.iter()
.filter(|id| {
cache.order(id).is_some_and(|linked| {
linked.strategy_id() == order.strategy_id()
&& Self::position_id_for_order(&cache, &linked).or_else(|| {
self.position_for_order_in_cache(&cache, &linked)
.map(|position| position.id)
}) == position_id
})
})
.copied()
.collect()
}
fn reduce_only_order_ids(&self, position_id: PositionId) -> Vec<ClientOrderId> {
if !self.config.use_reduce_only {
return Vec::new();
}
let cache = self.cache.borrow();
let mut order_ids = Vec::new();
for resting in self.core.iter_orders() {
let Some(order) = cache.order(&resting.client_order_id) else {
continue;
};
if !order.is_reduce_only() || !order.is_open() || !order.is_passive() {
continue;
}
let matches_position = match cache.position_id(&resting.client_order_id) {
Some(id) => *id == position_id,
None => self
.position_for_order_in_cache(&cache, &order)
.is_some_and(|position| position.id == position_id),
};
if matches_position {
order_ids.push(resting.client_order_id);
}
}
order_ids.sort_unstable();
order_ids
}
fn sync_reduce_only_orders(
&mut self,
filled_order: &OrderAny,
position: Option<&Position>,
order_ids: &[ClientOrderId],
) -> anyhow::Result<()> {
for &client_order_id in order_ids {
if client_order_id == filled_order.client_order_id()
|| !self.core.order_exists(client_order_id)
{
continue;
}
let Some(mut order) = self.order_snapshot(client_order_id) else {
continue;
};
if !order.is_reduce_only() || order.is_closed() || !order.is_passive() {
continue;
}
let position = {
let cache = self.cache.borrow();
position.map_or_else(
|| self.position_for_order_in_cache(&cache, &order),
|position| {
Some(cache.position(&position.id).map_or_else(
|| position.clone_without_events(),
|position| position.clone_without_events(),
))
},
)
};
let remaining = self.position_quantity_remaining(&order, position.as_ref())?;
if remaining.is_zero() {
self.cancel_reduce_only_order(&order, filled_order.client_order_id())?;
continue;
}
let leaves = self.parent_capped_leaves(&order, remaining);
let target = order.filled_qty().checked_add(leaves).ok_or_else(|| {
anyhow::anyhow!("Reduce-only quantity overflow for order {client_order_id}")
})?;
if self.resize_order(&mut order, target) {
if target == order.filled_qty() {
self.cancel_reduce_only_order(&order, filled_order.client_order_id())?;
} else if self.config.support_contingent_orders
&& order.contingency_type() == Some(ContingencyType::Ouo)
{
self.sync_ouo_leaves(&order, leaves, filled_order.client_order_id())?;
}
}
}
Ok(())
}
fn cancel_reduce_only_order(
&mut self,
order: &OrderAny,
filled_order_id: ClientOrderId,
) -> anyhow::Result<()> {
let propagate = self.config.support_contingent_orders
&& order.contingency_type() == Some(ContingencyType::Ouo);
self.cancel_order(order, Some(!propagate));
if propagate {
self.sync_ouo_leaves(
order,
Quantity::zero(order.quantity().precision),
filled_order_id,
)?;
}
Ok(())
}
fn parent_capped_leaves(&self, order: &OrderAny, leaves: Quantity) -> Quantity {
let parent = if self.config.support_contingent_orders {
order
.parent_order_id()
.and_then(|id| self.order_snapshot(id))
} else {
None
};
parent.map_or(leaves, |parent| {
min(
leaves,
parent.filled_qty().saturating_sub(order.filled_qty()),
)
})
}
fn sync_ouo_leaves(
&mut self,
order: &OrderAny,
leaves: Quantity,
filled_order_id: ClientOrderId,
) -> anyhow::Result<()> {
for &client_order_id in order.linked_order_ids().into_iter().flatten() {
if client_order_id == filled_order_id || !self.core.order_exists(client_order_id) {
continue;
}
let Some(mut sibling) = self.order_snapshot(client_order_id) else {
continue;
};
if sibling.is_closed() || sibling.is_active_local() || !sibling.is_passive() {
continue;
}
if leaves.is_zero() {
self.cancel_order(&sibling, Some(false));
continue;
}
if !sibling.is_open() {
continue;
}
let leaves = self.parent_capped_leaves(&sibling, leaves);
let target = sibling.filled_qty().checked_add(leaves).ok_or_else(|| {
anyhow::anyhow!("OUO quantity overflow for order {client_order_id}")
})?;
self.resize_order(&mut sibling, target);
if leaves.is_zero() {
self.cancel_order(&sibling, Some(false));
}
}
Ok(())
}
fn resize_order(&self, order: &mut OrderAny, target: Quantity) -> bool {
let quote = order.is_quote_quantity() && !self.instrument.is_inverse();
if order.quantity() == target && !quote {
return false;
}
if quote {
order.set_is_quote_quantity(false);
}
self.generate_order_updated(order, target, order.price(), order.trigger_price(), None);
true
}
fn fee_underlying_price(&self) -> CorrectnessResult<Option<Price>> {
if !matches!(
self.instrument,
InstrumentAny::CryptoOption(_) | InstrumentAny::OptionContract(_)
) {
return Ok(None);
}
let Some(underlying) = self.instrument.underlying() else {
return Ok(None);
};
let underlying_id = InstrumentId::from(format!("{underlying}.{}", self.venue).as_str());
let instrument_id = self.instrument.id();
let cache = self.cache.borrow();
if let Some(price) = cache
.price(&underlying_id, PriceType::Last)
.or_else(|| cache.price(&underlying_id, PriceType::Mark))
.or_else(|| cache.price(&underlying_id, PriceType::Mid))
{
return Ok(Some(price));
}
cache
.option_greeks(&instrument_id)
.and_then(|greeks| greeks.underlying_price)
.map(|price| Price::new_checked(price, FIXED_PRECISION))
.transpose()
}
fn cached_order_is_closed(&self, client_order_id: ClientOrderId) -> bool {
self.cache
.borrow()
.order(&client_order_id)
.is_none_or(|order| order.is_closed())
}
fn purge_cached_filled_qty_if_closed(&self, client_order_id: ClientOrderId) {
if self.cached_order_is_closed(client_order_id) {
self.orders
.cached_filled_qty
.borrow_mut()
.swap_remove(&client_order_id);
}
}
fn has_nothing_to_match(&self, timestamp_ns: UnixNanos) -> bool {
self.core.iter_orders().next().is_none()
&& self.post_match_order_ids.is_empty()
&& self.pending_fills.is_empty()
&& self.orders.cached_filled_qty.borrow().is_empty()
&& self.orders.pending_updates.borrow().is_empty()
&& self.orders.rejected_oto_parent_ids.borrow().is_empty()
&& self.orders.closed_oto_order_ids.borrow().is_empty()
&& self.target_bid.is_none()
&& self.target_ask.is_none()
&& self.target_last.is_none()
&& self.instrument_close.is_none()
&& self
.instrument
.expiration_ns()
.is_none_or(|expiration_ns| timestamp_ns < expiration_ns)
}
fn purge_closed_cached_filled_qty(&self) {
let client_order_ids: Vec<ClientOrderId> = self
.orders
.cached_filled_qty
.borrow()
.keys()
.copied()
.collect();
for client_order_id in client_order_ids {
self.purge_cached_filled_qty_if_closed(client_order_id);
}
}
fn update_limit_order(
&mut self,
order: &OrderAny,
quantity: Quantity,
price: Price,
) -> ModifyOutcome {
if self.core.is_limit_matched(order.order_side(), price) {
if order.is_post_only() {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(format!(
"POST_ONLY {} {} order with new limit px of {} would have been a TAKER: bid={}, ask={}",
order.order_type(),
order.order_side(),
price,
self.core.bid.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core.ask.map_or_else(|| "None".to_string(), |p| p.to_string())
).as_str()),
order.venue_order_id(),
order.account_id(),
);
return ModifyOutcome::Rejected;
}
self.generate_order_updated(order, quantity, Some(price), None, None);
let client_order_id = order.client_order_id();
if let Some(mut order) = self.cache.borrow_mut().order_mut(&client_order_id) {
order.set_liquidity_side(LiquiditySide::Taker);
}
self.fill_limit_order(client_order_id);
return ModifyOutcome::Applied;
}
self.generate_order_updated(order, quantity, Some(price), None, None);
ModifyOutcome::Applied
}
fn update_stop_market_order(
&self,
order: &OrderAny,
quantity: Quantity,
trigger_price: Price,
) -> ModifyOutcome {
if self.core.is_stop_matched_with_trigger_type(
order.order_side(),
trigger_price,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(
format!(
"{} {} order new stop px of {} was in the market: bid={}, ask={}",
order.order_type(),
order.order_side(),
trigger_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.as_str(),
),
order.venue_order_id(),
order.account_id(),
);
return ModifyOutcome::Rejected;
}
self.generate_order_updated(order, quantity, None, Some(trigger_price), None);
ModifyOutcome::Applied
}
fn update_stop_limit_order(
&mut self,
order: &OrderAny,
quantity: Quantity,
price: Price,
trigger_price: Price,
) -> ModifyOutcome {
if order.is_triggered().is_some_and(|t| t) {
if self.core.is_limit_matched(order.order_side(), price) {
return self.update_limit_order(order, quantity, price);
}
} else {
if self.core.is_stop_matched_with_trigger_type(
order.order_side(),
trigger_price,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(
format!(
"{} {} order new stop px of {} was in the market: bid={}, ask={}",
order.order_type(),
order.order_side(),
trigger_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.as_str(),
),
order.venue_order_id(),
order.account_id(),
);
return ModifyOutcome::Rejected;
}
}
self.generate_order_updated(order, quantity, Some(price), Some(trigger_price), None);
ModifyOutcome::Applied
}
fn update_market_if_touched_order(
&self,
order: &OrderAny,
quantity: Quantity,
trigger_price: Price,
) -> ModifyOutcome {
if self.core.is_touch_triggered_with_trigger_type(
order.order_side(),
trigger_price,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(
format!(
"{} {} order new trigger px of {} was in the market: bid={}, ask={}",
order.order_type(),
order.order_side(),
trigger_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.as_str(),
),
order.venue_order_id(),
order.account_id(),
);
return ModifyOutcome::Rejected;
}
self.generate_order_updated(order, quantity, None, Some(trigger_price), None);
ModifyOutcome::Applied
}
fn update_limit_if_touched_order(
&mut self,
order: &OrderAny,
quantity: Quantity,
price: Price,
trigger_price: Price,
) -> ModifyOutcome {
if order.is_triggered().is_some_and(|t| t) {
if self.core.is_limit_matched(order.order_side(), price) {
return self.update_limit_order(order, quantity, price);
}
} else {
if self.core.is_touch_triggered_with_trigger_type(
order.order_side(),
trigger_price,
order.trigger_type().unwrap_or(TriggerType::Default),
) {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(
format!(
"{} {} order new trigger px of {} was in the market: bid={}, ask={}",
order.order_type(),
order.order_side(),
trigger_price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.as_str(),
),
order.venue_order_id(),
order.account_id(),
);
return ModifyOutcome::Rejected;
}
}
self.generate_order_updated(order, quantity, Some(price), Some(trigger_price), None);
ModifyOutcome::Applied
}
fn update_trailing_stop_order(&self, order: &OrderAny) {
let (new_trigger_price, new_price) = match trailing_stop_calculate(
self.instrument.price_increment(),
order.trigger_price(),
order,
self.core.bid,
self.core.ask,
if order.trigger_type() == Some(TriggerType::MarkPrice) {
self.core.mark
} else {
self.core.last
},
) {
Ok(prices) => prices,
Err(e) => {
log::debug!("Cannot calculate trailing-stop update: {e}");
return;
}
};
if new_trigger_price.is_none() && new_price.is_none() {
return;
}
self.generate_order_updated(order, order.quantity(), new_price, new_trigger_price, None);
}
fn accept_order(&mut self, order: &mut OrderAny) {
if order.is_closed() {
return;
}
if order.status() != OrderStatus::Accepted {
let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
let event = self.create_order_accepted(order, venue_order_id);
if let Err(e) = order.apply(event.clone()) {
log::warn!(
"Skipping local apply of accepted event for {}: {e}",
order.client_order_id(),
);
}
self.dispatch_order_event(event);
if matches!(
order.order_type(),
OrderType::TrailingStopLimit | OrderType::TrailingStopMarket
) && order.trigger_price().is_none()
&& self.maybe_activate_trailing_stop(
order,
self.core.bid,
self.core.ask,
self.core.last,
)
{
self.update_trailing_stop_order(order);
}
}
let match_info = Self::matching_core_entry(order);
self.track_post_match_order(order);
self.core.add_order(match_info);
}
fn track_post_match_order(&mut self, order: &OrderAny) {
self.post_match_order_ids.insert(order.client_order_id());
}
fn delete_core_order(&mut self, client_order_id: ClientOrderId) {
self.post_match_order_ids.swap_remove(&client_order_id);
let _ = self.core.delete_order(client_order_id);
}
fn requires_post_match_maintenance(order: &OrderAny) -> bool {
order.expire_time().is_some()
|| matches!(
order.order_type(),
OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
)
}
fn matching_core_entry(order: &OrderAny) -> RestingOrder {
let is_limit_style_stop = matches!(
order.order_type(),
OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit
);
let is_triggered = order.is_triggered().is_some_and(|triggered| triggered);
let triggered_limit_style = is_limit_style_stop && is_triggered;
let limit_price = if is_limit_style_stop && !is_triggered && order.trigger_price().is_none()
{
None
} else {
order.price()
};
RestingOrder::new_with_trigger_type(
order.client_order_id(),
order.order_side(),
order.order_type(),
Some(order.trigger_type().unwrap_or(TriggerType::Default)),
if triggered_limit_style {
None
} else {
order.trigger_price()
},
limit_price,
match order {
OrderAny::TrailingStopMarket(o) => o.is_activated,
OrderAny::TrailingStopLimit(o) => o.is_activated,
_ => true,
},
)
}
fn expire_order(&mut self, order: &OrderAny) {
self.remove_queue_position(order.client_order_id());
if self.config.support_contingent_orders && order.contingency_type().is_some() {
self.cancel_contingent_orders(order, &[]);
}
self.generate_order_expired(order);
}
fn cancel_order(&mut self, order: &OrderAny, cancel_contingencies: Option<bool>) {
self.cancel_order_excluding(order, cancel_contingencies, &[]);
}
fn cancel_order_excluding(
&mut self,
order: &OrderAny,
cancel_contingencies: Option<bool>,
excluded: &[ClientOrderId],
) {
if self.inflight_orders.contains(order.client_order_id()) {
return;
}
let cancel_contingencies = cancel_contingencies.unwrap_or(true);
if order.instrument_id() != self.instrument.id() && order.parent_order_id().is_some() {
let mut commands = self.orders.commands.borrow_mut();
if !commands.iter().any(|command| matches!(command,
OtoCommand::Cancel { client_order_id, .. } if *client_order_id == order.client_order_id()))
{
commands.push_front(OtoCommand::Cancel {
instrument_id: order.instrument_id(), client_order_id: order.client_order_id(),
cancel_contingencies, excluded: excluded.to_vec(),
});
}
return;
}
if order.is_active_local()
&& !matches!(
(order.status(), order.order_type()),
(
OrderStatus::Initialized | OrderStatus::Released,
OrderType::Market
)
)
{
log::error!(
"Cannot cancel an order with {} from the matching engine",
order.status()
);
return;
}
if self.core.order_exists(order.client_order_id()) {
self.delete_core_order(order.client_order_id());
}
self.orders
.pending_oto_order_ids
.borrow_mut()
.swap_remove(&order.client_order_id());
self.remove_queue_position(order.client_order_id());
let venue_order_id = self.ids_generator.get_venue_order_id(order).unwrap();
self.generate_order_canceled(order, venue_order_id);
if self.config.support_contingent_orders
&& order.contingency_type().is_some()
&& cancel_contingencies
{
self.cancel_contingent_orders(order, excluded);
}
self.purge_cached_filled_qty_if_closed(order.client_order_id());
}
fn update_order(
&mut self,
order: &OrderAny,
quantity: Option<Quantity>,
price: Option<Price>,
trigger_price: Option<Price>,
update_contingencies: Option<bool>,
) -> bool {
if self.inflight_orders.contains(order.client_order_id()) {
return false;
}
let update_contingencies = update_contingencies.unwrap_or(true);
let quantity = quantity.unwrap_or(order.quantity());
let size_prec = self.instrument.size_precision();
let instrument_id = self.instrument.id();
if !order_precision_valid(quantity.precision, size_prec) {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(&format!(
"Invalid update quantity precision {}, expected {size_prec} for {instrument_id}",
quantity.precision
)),
order.venue_order_id(),
order.account_id(),
);
return false;
}
let price_rejection = price
.filter(|px| Some(*px) != order.price())
.and_then(|px| self.update_price_rejection("price", px))
.or_else(|| {
trigger_price
.filter(|tp| Some(*tp) != order.trigger_price())
.and_then(|tp| self.update_price_rejection("trigger_price", tp))
});
if let Some(reason) = price_rejection {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(&reason),
order.venue_order_id(),
order.account_id(),
);
return false;
}
let filled_qty = self.engine_filled_qty(order);
if quantity < filled_qty {
self.generate_order_modify_rejected(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
Ustr::from(&format!(
"Cannot reduce order quantity {quantity} below filled quantity {filled_qty}",
)),
order.venue_order_id(),
order.account_id(),
);
return false;
}
let outcome = match order {
OrderAny::Limit(_) | OrderAny::MarketToLimit(_) => {
let price = price.unwrap_or(order.price().unwrap());
self.update_limit_order(order, quantity, price)
}
OrderAny::StopMarket(_) => {
let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
self.update_stop_market_order(order, quantity, trigger_price)
}
OrderAny::StopLimit(_) => {
let price = price.unwrap_or(order.price().unwrap());
let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
self.update_stop_limit_order(order, quantity, price, trigger_price)
}
OrderAny::MarketIfTouched(_) => {
let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
self.update_market_if_touched_order(order, quantity, trigger_price)
}
OrderAny::LimitIfTouched(_) => {
let price = price.unwrap_or(order.price().unwrap());
let trigger_price = trigger_price.unwrap_or(order.trigger_price().unwrap());
self.update_limit_if_touched_order(order, quantity, price, trigger_price)
}
OrderAny::TrailingStopMarket(inner) => match trigger_price.or(order.trigger_price()) {
Some(trigger_price)
if inner.is_activated || self.trailing_stop_activates_now(order) =>
{
self.update_stop_market_order(order, quantity, trigger_price)
}
trigger_price => {
self.generate_order_updated(order, quantity, None, trigger_price, None);
ModifyOutcome::Applied
}
},
OrderAny::TrailingStopLimit(inner) => match (
price.or(order.price()),
trigger_price.or(order.trigger_price()),
) {
(Some(price), Some(trigger_price))
if inner.is_activated || self.trailing_stop_activates_now(order) =>
{
self.update_stop_limit_order(order, quantity, price, trigger_price)
}
(price, trigger_price) => {
self.generate_order_updated(order, quantity, price, trigger_price, None);
ModifyOutcome::Applied
}
},
_ => {
panic!(
"Unsupported order type {} for update_order",
order.order_type()
);
}
};
if outcome == ModifyOutcome::Rejected {
return false;
}
if self.config.support_contingent_orders
&& order.contingency_type().is_some()
&& update_contingencies
{
self.update_contingent_order(order, quantity);
}
let new_leaves_qty = quantity.saturating_sub(filled_qty);
if new_leaves_qty.is_zero() {
self.cancel_order(order, Some(false));
return true;
}
true
}
fn update_price_rejection(&self, field: &str, price: Price) -> Option<String> {
let price_precision = self.instrument.price_precision();
let instrument_id = self.instrument.id();
if !order_precision_valid(price.precision, price_precision) {
return Some(format!(
"Invalid update {field} precision {}, expected {price_precision} for {instrument_id}",
price.precision
));
}
let price_increment = self.instrument.price_increment();
if !Self::price_matches_tick(price, price_increment) {
return Some(format!(
"Invalid update {field} {price}, not a multiple of {instrument_id} price increment {price_increment}"
));
}
None
}
pub fn trigger_stop_order(&mut self, client_order_id: ClientOrderId) {
let order = match self.order_snapshot(client_order_id) {
Some(order) => order,
None => {
log::error!(
"Cannot trigger stop order: order {client_order_id} not found in cache"
);
return;
}
};
if order.is_closed() {
log::debug!("Cannot trigger stop order: {client_order_id} already closed");
return;
}
match order.order_type() {
OrderType::StopLimit | OrderType::LimitIfTouched | OrderType::TrailingStopLimit => {
self.trigger_limit_style_stop_order(client_order_id, order);
}
OrderType::StopMarket | OrderType::MarketIfTouched | OrderType::TrailingStopMarket => {
self.fill_market_order(client_order_id);
}
_ => {
log::error!(
"Cannot trigger stop order: invalid order type {}",
order.order_type()
);
}
}
}
fn trigger_limit_style_stop_order(&mut self, client_order_id: ClientOrderId, order: OrderAny) {
if order.is_triggered().is_some_and(|triggered| triggered) {
let liquidity_side = match (order.price(), order.trigger_price()) {
(Some(price), Some(trigger_price)) => self.determine_triggered_limit_liquidity(
order.order_side(),
price,
trigger_price,
order.trigger_type(),
),
_ => LiquiditySide::Maker,
};
if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id)
&& !matches!(
cached_order.liquidity_side(),
Some(LiquiditySide::Maker | LiquiditySide::Taker)
)
{
cached_order.set_liquidity_side(liquidity_side);
}
self.fill_limit_order(client_order_id);
return;
}
let event = self.create_order_triggered(&order);
let order = match self.cache.borrow_mut().update_order(&event) {
Ok(order) => order,
Err(e) => {
log::debug!(
"Failed to apply triggered event for {} before fill: {e}",
order.client_order_id(),
);
order
}
};
let order = self.order_snapshot(client_order_id).unwrap_or(order);
self.dispatch_order_event(event);
let trigger_price = order
.trigger_price()
.expect("Limit-style stop order must have a trigger price");
let price = order
.price()
.expect("Limit-style stop order must have a price");
let liquidity_side = self.determine_triggered_limit_liquidity(
order.order_side(),
price,
trigger_price,
order.trigger_type(),
);
let maker_inside = match order.order_side() {
OrderSide::Buy => self
.core
.ask
.is_some_and(|ask| liquidity_side == LiquiditySide::Maker && price > ask),
OrderSide::Sell => self
.core
.bid
.is_some_and(|bid| liquidity_side == LiquiditySide::Maker && price < bid),
};
if maker_inside {
if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
cached_order.set_liquidity_side(LiquiditySide::Maker);
}
self.resync_core_entry(client_order_id);
self.fill_limit_order(client_order_id);
return;
}
if self.core.is_limit_matched(order.order_side(), price) {
if order.is_post_only() {
self.delete_core_order(client_order_id);
self.orders
.cached_filled_qty
.borrow_mut()
.swap_remove(&client_order_id);
let event = self.create_order_rejected(
&order,
format!(
"POST_ONLY {} {} order limit px of {} would have been a TAKER: bid={}, ask={}",
order.order_type(),
order.order_side(),
price,
self.core
.bid
.map_or_else(|| "None".to_string(), |p| p.to_string()),
self.core
.ask
.map_or_else(|| "None".to_string(), |p| p.to_string())
)
.into(),
);
if let Err(e) = self.cache.borrow_mut().update_order(&event) {
log::debug!(
"Failed to apply rejected event for {} after post-only trigger: {e}",
order.client_order_id(),
);
}
self.dispatch_order_event(event);
return;
}
if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
cached_order.set_liquidity_side(LiquiditySide::Taker);
}
self.resync_core_entry(client_order_id);
self.fill_limit_order(client_order_id);
return;
}
if let Some(mut cached_order) = self.cache.borrow_mut().order_mut(&client_order_id) {
cached_order.set_liquidity_side(liquidity_side);
}
self.resync_core_entry(client_order_id);
}
fn determine_triggered_limit_liquidity(
&self,
side: OrderSide,
price: Price,
trigger_price: Price,
trigger_type: Option<TriggerType>,
) -> LiquiditySide {
if trigger_type == Some(TriggerType::MarkPrice) {
return if self.core.is_limit_matched(side, price) {
LiquiditySide::Taker
} else {
LiquiditySide::Maker
};
}
if (side == OrderSide::Buy && trigger_price > price)
|| (side == OrderSide::Sell && trigger_price < price)
{
LiquiditySide::Maker
} else {
LiquiditySide::Taker
}
}
fn update_contingent_order(&mut self, order: &OrderAny, parent_quantity: Quantity) {
log::debug!(
"Updating contingent orders from {}",
order.client_order_id()
);
if order.contingency_type() == Some(ContingencyType::Oto) {
self.update_oto_children(order, Some(parent_quantity), &[]);
return;
}
if let Some(linked_order_ids) = order.linked_order_ids() {
let parent_filled_qty = self.engine_filled_qty(order);
let parent_leaves_qty = parent_quantity.saturating_sub(parent_filled_qty);
for client_order_id in linked_order_ids {
let child_order = match self.order_snapshot(*client_order_id) {
Some(order) => order,
None => panic!("Order {client_order_id} not found in cache."),
};
if child_order.is_active_local() {
continue;
}
let child_filled_qty = self.engine_filled_qty(&child_order);
if child_filled_qty >= parent_leaves_qty {
self.cancel_order(&child_order, Some(false));
} else {
self.update_contingent_quantity(
&child_order,
parent_leaves_qty,
order.is_quote_quantity() && !self.instrument.is_inverse(),
);
}
}
}
}
fn update_contingent_quantity(
&mut self,
order: &OrderAny,
quantity: Quantity,
is_quote_quantity: bool,
) {
if self.contingent_order_is_closed(order)
|| self.inflight_orders.contains(order.client_order_id())
{
return;
}
let quantity = self.quantity_for_order(order, quantity);
if quantity <= self.engine_filled_qty(order) {
self.cancel_order(order, Some(false));
return;
}
let clear_quote = order.is_quote_quantity()
&& !is_quote_quantity
&& if order.instrument_id() == self.instrument.id() {
!self.instrument.is_inverse()
} else {
self.cache
.borrow()
.instrument(&order.instrument_id())
.is_some_and(|instrument| !instrument.is_inverse())
};
if quantity == order.leaves_qty() && !clear_quote {
return;
}
let mut converted;
let order = if clear_quote {
converted = order.clone();
converted.set_is_quote_quantity(false);
&converted
} else {
order
};
let price = order.price();
let trigger_price = order.trigger_price();
if order.instrument_id() == self.instrument.id() {
self.update_order(order, Some(quantity), price, trigger_price, Some(false));
} else {
self.generate_order_updated(order, quantity, price, trigger_price, None);
}
}
fn apply_deferred_submission(
&mut self,
order: &mut OrderAny,
accepted_elsewhere: bool,
) -> anyhow::Result<()> {
let client_order_id = order.client_order_id();
let accepted = self.core.order_exists(client_order_id) || accepted_elsewhere;
if !order.is_active_local()
|| !(accepted
|| self
.orders
.pending_oto_order_ids
.borrow()
.contains(&client_order_id))
{
return Ok(());
}
let account_id = match order.account_id() {
Some(account_id) => account_id,
None => *self
.account_ids
.get(&order.trader_id())
.ok_or_else(|| anyhow::anyhow!("No account ID for {}", order.trader_id()))?,
};
let event = self.create_order_submitted(order, account_id);
order.apply(event)?;
if accepted {
let venue_order_id = self.ids_generator.get_venue_order_id(order)?;
let event = self.create_order_accepted(order, venue_order_id);
order.apply(event)?;
}
Ok(())
}
fn works_in_another_engine(&self, order: &OrderAny) -> bool {
order.instrument_id() != self.instrument.id()
&& (self
.orders
.released_oto_order_ids
.borrow()
.contains(&order.client_order_id())
|| self
.cache
.borrow()
.venue_order_id(&order.client_order_id())
.is_some())
}
fn update_oto_children(
&mut self,
parent: &OrderAny,
parent_quantity: Option<Quantity>,
excluded: &[ClientOrderId],
) {
let Some(ids) = parent.linked_order_ids() else {
return;
};
let snapshot = self
.order_snapshot(parent.client_order_id())
.unwrap_or_else(|| parent.clone());
let filled_qty = self.engine_filled_qty(&snapshot);
let quantity = parent_quantity.unwrap_or(filled_qty);
let sizing = OtoSizing {
quantity,
filled_qty,
quote: snapshot.is_quote_quantity() && !self.instrument.is_inverse(),
reduced_to_filled: parent_quantity.is_some_and(|q| q <= filled_qty),
grew: quantity > parent.quantity(),
};
let mut released = false;
for id in ids {
if excluded.contains(id) {
continue;
}
released |= self.update_oto_child(parent, *id, ids, &sizing);
}
if released {
let not_resting: Vec<_> = ids
.iter()
.filter(|id| {
excluded.contains(*id)
|| (!self.core.order_exists(**id)
&& !self
.order_snapshot(**id)
.is_some_and(|order| self.works_in_another_engine(&order)))
})
.copied()
.collect();
self.update_oto_children(parent, Some(quantity), ¬_resting);
}
}
fn update_oto_child(
&mut self,
parent: &OrderAny,
id: ClientOrderId,
ids: &[ClientOrderId],
sizing: &OtoSizing,
) -> bool {
let mut child = self
.order_snapshot(id)
.unwrap_or_else(|| panic!("Cannot find contingent order for {id}"));
let elsewhere = self.works_in_another_engine(&child);
let accepted_elsewhere = elsewhere && self.cache.borrow().venue_order_id(&id).is_some();
if let Err(e) = self.apply_deferred_submission(&mut child, accepted_elsewhere) {
log::error!("Cannot update contingent order {id}: {e}");
return false;
}
if child.is_active_local()
|| self.contingent_order_is_closed(&child)
|| self.inflight_orders.contains(id)
{
return false;
}
let remaining = self.oto_child_remaining(&child, ids, sizing.quantity);
let released = self.core.order_exists(id) || elsewhere;
if !released && !self.config.oto_full_trigger && !sizing.filled_qty.is_zero() {
return false;
}
let release = !released && sizing.reduced_to_filled && self.config.oto_full_trigger;
let parent_closed = sizing.quantity <= sizing.filled_qty;
if parent_closed && ((!released && !release) || remaining.is_zero()) {
self.cancel_order(&child, Some(false));
return false;
}
if remaining.is_zero() {
return false;
}
self.resize_oto_child(&mut child, sizing, remaining);
if !release {
return false;
}
self.release_oto_child_after_resize(parent, id, sizing.quantity)
}
fn contingent_order_is_closed(&self, order: &OrderAny) -> bool {
order.is_closed()
|| order.leaves_qty().is_zero()
|| self
.orders
.closed_oto_order_ids
.borrow()
.contains(&order.client_order_id())
}
fn oto_child_remaining(
&self,
child: &OrderAny,
ids: &[ClientOrderId],
quantity: Quantity,
) -> Quantity {
let filled = ids
.iter()
.filter_map(|id| self.order_snapshot(*id))
.fold(Quantity::zero(quantity.precision), |acc, order| {
acc + self.engine_filled_qty(&order)
});
self.quantity_for_order(child, quantity.saturating_sub(filled))
}
fn quantity_for_order(&self, order: &OrderAny, quantity: Quantity) -> Quantity {
if order.instrument_id() == self.instrument.id() {
let precision = self.instrument.size_precision();
if order_precision_valid(quantity.precision, precision) {
return quantity;
}
let rounded = quantity
.as_decimal()
.round_dp_with_strategy(u32::from(precision), RoundingStrategy::ToZero);
return Quantity::from_decimal_dp(rounded, precision)
.unwrap_or_else(|_| Quantity::zero(precision));
}
let cache = self.cache.borrow();
cache
.instrument(&order.instrument_id())
.map_or(quantity, |instrument| {
if quantity.precision <= FIXED_PRECISION
&& quantity.precision == instrument.size_precision()
&& Self::quantity_matches_precision(quantity, 0)
{
return quantity;
}
instrument
.try_make_qty_from_decimal(quantity.as_decimal(), Some(true))
.unwrap_or_else(|_| Quantity::zero(instrument.size_precision()))
})
}
fn resize_oto_child(&self, child: &mut OrderAny, sizing: &OtoSizing, remaining: Quantity) {
let quote = child.is_quote_quantity();
let other_instrument = child.instrument_id() != self.instrument.id();
if !self.convert_to_parent_units(child, sizing.quote, other_instrument) {
return;
}
let converted = quote != child.is_quote_quantity();
let parent_closed = sizing.quantity <= sizing.filled_qty;
let mut quantity = self.engine_filled_qty(child) + remaining;
if parent_closed || !sizing.grew {
quantity = quantity.min(child.quantity());
}
if quantity != child.quantity() || (parent_closed && converted) {
self.generate_order_updated(
child,
quantity,
child.price(),
child.trigger_price(),
None,
);
}
}
fn release_oto_child_after_resize(
&mut self,
parent: &OrderAny,
id: ClientOrderId,
quantity: Quantity,
) -> bool {
let account_id = parent
.account_id()
.or_else(|| self.account_ids.get(&parent.trader_id()).copied());
match (self.order_snapshot(id), account_id) {
(Some(mut child), Some(account_id)) => {
self.release_oto_child(&mut child, account_id, parent, Some(quantity));
true
}
_ => {
log::error!("Cannot release OTO order {id}");
false
}
}
}
fn release_oto_child(
&mut self,
child: &mut OrderAny,
account_id: AccountId,
parent: &OrderAny,
parent_quantity: Option<Quantity>,
) {
if child.instrument_id() == self.instrument.id() {
self.process_order(child, account_id);
} else if self
.orders
.released_oto_order_ids
.borrow_mut()
.insert(child.client_order_id())
{
self.orders
.commands
.borrow_mut()
.push_back(OtoCommand::Release {
instrument_id: child.instrument_id(),
client_order_id: child.client_order_id(),
account_id,
parent_instrument_id: self.instrument.id(),
parent_order_id: parent.client_order_id(),
parent_quantity,
});
}
}
fn convert_to_parent_units(
&self,
child: &mut OrderAny,
parent_quote: bool,
other_instrument: bool,
) -> bool {
let cache = self.cache.borrow();
let instrument = cache
.instrument(&child.instrument_id())
.filter(|_| other_instrument)
.unwrap_or(&self.instrument);
let child_quote = child.is_quote_quantity() && !instrument.is_inverse();
if child_quote == parent_quote {
return true;
}
if parent_quote {
return false;
}
let opposing_best = match child.order_side() {
OrderSide::Buy => self.core.ask,
OrderSide::Sell => self.core.bid,
};
let reference_price = child
.price()
.or(child.trigger_price())
.or(opposing_best.filter(|_| !other_instrument));
let base_quantity = reference_price.and_then(|price| {
instrument
.try_calculate_base_quantity(child.quantity(), price)
.ok()
});
let Some(base_quantity) = base_quantity else {
return false;
};
child.set_quantity(base_quantity);
child.set_is_quote_quantity(false);
true
}
fn engine_filled_qty(&self, order: &OrderAny) -> Quantity {
self.orders
.cached_filled_qty
.borrow()
.get(&order.client_order_id())
.copied()
.unwrap_or(order.filled_qty())
}
fn cancel_contingent_orders(&mut self, order: &OrderAny, excluded: &[ClientOrderId]) {
if order.contingency_type() == Some(ContingencyType::Oto) {
self.update_oto_children(order, None, excluded);
return;
}
if let Some(linked_order_ids) = order.linked_order_ids() {
for client_order_id in linked_order_ids {
if excluded.contains(client_order_id) {
continue;
}
let mut contingent_order = match self.order_snapshot(*client_order_id) {
Some(order) => order,
None => panic!("Cannot find contingent order for {client_order_id}"),
};
if let Err(e) = self.apply_deferred_submission(&mut contingent_order, false) {
log::error!("Cannot cancel contingent order {client_order_id}: {e}");
continue;
}
if contingent_order.is_active_local() {
continue;
}
if !self.contingent_order_is_closed(&contingent_order) {
self.cancel_order(&contingent_order, Some(false));
}
}
}
}
fn create_order_submitted(&self, order: &OrderAny, account_id: AccountId) -> OrderEventAny {
let ts_now = self.clock.borrow().timestamp_ns();
OrderEventAny::Submitted(OrderSubmitted::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
account_id,
UUID4::new(),
ts_now,
ts_now,
))
}
fn generate_order_submitted(&self, order: &OrderAny, account_id: AccountId) {
let event = self.create_order_submitted(order, account_id);
self.dispatch_order_event(event);
}
fn create_order_rejected(&self, order: &OrderAny, reason: Ustr) -> OrderEventAny {
let ts_now = self.clock.borrow().timestamp_ns();
let account_id = order
.account_id()
.unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
let due_post_only = reason.starts_with("POST_ONLY");
OrderEventAny::Rejected(OrderRejected::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
account_id,
reason,
UUID4::new(),
ts_now,
ts_now,
false,
due_post_only,
))
}
fn generate_order_rejected(&self, order: &OrderAny, reason: Ustr) {
if order.contingency_type() == Some(ContingencyType::Oto) {
self.orders
.rejected_oto_parent_ids
.borrow_mut()
.insert(order.client_order_id());
}
let event = self.create_order_rejected(order, reason);
self.dispatch_oto_terminal(order, event);
}
fn publish_order_initialized(&self, order: &OrderAny) {
let event = OrderEventAny::Initialized(order.init_event().clone());
let topic = msgbus::switchboard::get_event_order_topic(order.strategy_id());
msgbus::publish_order_event(topic, &event);
}
fn create_order_accepted(
&self,
order: &OrderAny,
venue_order_id: VenueOrderId,
) -> OrderEventAny {
let ts_now = self.clock.borrow().timestamp_ns();
let account_id = order
.account_id()
.unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
OrderEventAny::Accepted(OrderAccepted::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
venue_order_id,
account_id,
UUID4::new(),
ts_now,
ts_now,
false,
))
}
fn generate_order_accepted(&self, order: &OrderAny, venue_order_id: VenueOrderId) {
let event = self.create_order_accepted(order, venue_order_id);
self.dispatch_order_event(event);
}
#[expect(clippy::too_many_arguments)]
fn generate_order_modify_rejected(
&self,
trader_id: TraderId,
strategy_id: StrategyId,
instrument_id: InstrumentId,
client_order_id: ClientOrderId,
reason: Ustr,
venue_order_id: Option<VenueOrderId>,
account_id: Option<AccountId>,
) {
let ts_now = self.clock.borrow().timestamp_ns();
let event = OrderEventAny::ModifyRejected(OrderModifyRejected::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
reason,
UUID4::new(),
ts_now,
ts_now,
false,
venue_order_id,
account_id,
));
self.dispatch_order_event(event);
}
#[expect(clippy::too_many_arguments)]
fn generate_order_cancel_rejected(
&self,
trader_id: TraderId,
strategy_id: StrategyId,
account_id: AccountId,
instrument_id: InstrumentId,
client_order_id: ClientOrderId,
venue_order_id: Option<VenueOrderId>,
reason: Ustr,
) {
let ts_now = self.clock.borrow().timestamp_ns();
let event = OrderEventAny::CancelRejected(OrderCancelRejected::new(
trader_id,
strategy_id,
instrument_id,
client_order_id,
reason,
UUID4::new(),
ts_now,
ts_now,
false,
venue_order_id,
Some(account_id),
));
self.dispatch_order_event(event);
}
fn generate_order_updated(
&self,
order: &OrderAny,
quantity: Quantity,
price: Option<Price>,
trigger_price: Option<Price>,
protection_price: Option<Price>,
) {
let ts_now = self.clock.borrow().timestamp_ns();
let event = OrderUpdated::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
quantity,
UUID4::new(),
ts_now,
ts_now,
false,
order.venue_order_id(),
order.account_id(),
price,
trigger_price,
protection_price,
order.is_quote_quantity(),
);
self.orders
.pending_updates
.borrow_mut()
.entry(order.client_order_id())
.or_default()
.push(event);
self.dispatch_order_event(OrderEventAny::Updated(event));
}
fn generate_order_canceled(&self, order: &OrderAny, venue_order_id: VenueOrderId) {
let ts_now = self.clock.borrow().timestamp_ns();
let event = OrderEventAny::Canceled(OrderCanceled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
ts_now,
ts_now,
false,
Some(venue_order_id),
order.account_id(),
None,
));
self.dispatch_oto_terminal(order, event);
}
fn create_order_triggered(&self, order: &OrderAny) -> OrderEventAny {
let ts_now = self.clock.borrow().timestamp_ns();
OrderEventAny::Triggered(OrderTriggered::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
ts_now,
ts_now,
false,
order.venue_order_id(),
order.account_id(),
))
}
fn generate_order_expired(&self, order: &OrderAny) {
let ts_now = self.clock.borrow().timestamp_ns();
let event = OrderEventAny::Expired(OrderExpired::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
ts_now,
ts_now,
false,
order.venue_order_id(),
order.account_id(),
));
self.dispatch_oto_terminal(order, event);
}
fn dispatch_oto_terminal(&self, order: &OrderAny, event: OrderEventAny) {
if self.config.support_contingent_orders && order.parent_order_id().is_some() {
self.orders
.closed_oto_order_ids
.borrow_mut()
.insert(order.client_order_id());
}
self.dispatch_order_event(event);
if self.cached_order_is_closed(order.client_order_id()) {
self.orders
.closed_oto_order_ids
.borrow_mut()
.swap_remove(&order.client_order_id());
}
}
#[expect(clippy::too_many_arguments)]
fn generate_order_filled(
&mut self,
order: &OrderAny,
venue_order_id: VenueOrderId,
venue_position_id: Option<PositionId>,
last_qty: Quantity,
last_px: Price,
quote_currency: Currency,
commission: Money,
liquidity_side: LiquiditySide,
) {
debug_assert!(
last_qty <= order.quantity(),
"Fill quantity {last_qty} exceeds order quantity {order_qty} for {client_order_id}",
order_qty = order.quantity(),
client_order_id = order.client_order_id()
);
let ts_now = self.clock.borrow().timestamp_ns();
let account_id = order
.account_id()
.unwrap_or(self.account_ids.get(&order.trader_id()).unwrap().to_owned());
let fill = OrderFilled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
venue_order_id,
account_id,
self.ids_generator.generate_trade_id(ts_now),
order.order_side(),
order.order_type(),
last_qty,
last_px,
quote_currency,
liquidity_side,
UUID4::new(),
ts_now,
ts_now,
false,
venue_position_id,
Some(commission),
None,
);
self.record_pending_fill(&fill);
self.dispatch_order_event(OrderEventAny::Filled(fill));
}
fn record_pending_fill(&mut self, fill: &OrderFilled) {
if (!self.config.use_reduce_only
&& !self.restricts_short_selling()
&& !self.instrument.has_expiration())
|| self.instrument.is_spread()
{
return;
}
self.purge_applied_fills();
let cache = self.cache.borrow();
let position_id = cache
.position_id(&fill.client_order_id)
.copied()
.or(fill.position_id)
.or_else(|| {
(self.oms_type == OmsType::Netting).then(|| {
PositionId::new(format!("{}-{}", fill.instrument_id, fill.strategy_id))
})
});
let opening_trade_id = position_id.and_then(|id| {
cache
.position(&id)
.and_then(|position| position.events.first().map(|event| event.trade_id))
});
let mut quantity_change = if fill.order_side == OrderSide::Buy {
fill.last_qty.as_decimal()
} else {
-fill.last_qty.as_decimal()
};
if matches!(
self.instrument,
InstrumentAny::CurrencyPair(_) | InstrumentAny::TokenizedAsset(_)
) && let Some(commission) = fill.commission
&& Some(commission.currency) == self.instrument.base_currency()
{
quantity_change -= commission.as_decimal();
}
self.pending_fills.insert(
fill.trade_id,
PendingFill {
client_order_id: fill.client_order_id,
position_id,
opening_trade_id,
quantity_change,
},
);
}
}
struct OtoSizing {
quantity: Quantity,
filled_qty: Quantity,
quote: bool,
reduced_to_filled: bool,
grew: bool,
}
#[derive(Debug)]
struct PendingFill {
client_order_id: ClientOrderId,
position_id: Option<PositionId>,
opening_trade_id: Option<TradeId>,
quantity_change: Decimal,
}
#[derive(Debug)]
struct ConsumptionJournal {
order_side: OrderSide,
trade_consumption: QuantityRaw,
levels: Vec<(PriceRaw, Option<(QuantityRaw, QuantityRaw)>)>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ModifyOutcome {
Applied,
Rejected,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OrderMatchMode {
All,
StopTriggers(TriggerType),
}
#[derive(Debug)]
enum PostMatchOrderAction {
RemoveClosed,
Expire(OrderAny),
UpdateTrailing(OrderAny),
NoMaintenance,
}
fn order_precision_valid(actual: u8, expected: u8) -> bool {
actual <= expected && raw_scales_match(actual, expected)
}
fn post_match_order_action<F>(
order: &OrderAny,
support_gtd_orders: bool,
timestamp_ns: UnixNanos,
clone_order: F,
) -> PostMatchOrderAction
where
F: FnOnce(&OrderAny) -> OrderAny,
{
if order.is_closed() {
PostMatchOrderAction::RemoveClosed
} else if support_gtd_orders
&& order
.expire_time()
.is_some_and(|expire_ns| timestamp_ns >= expire_ns)
{
PostMatchOrderAction::Expire(clone_order(order))
} else if matches!(
order.order_type(),
OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
) {
PostMatchOrderAction::UpdateTrailing(clone_order(order))
} else {
PostMatchOrderAction::NoMaintenance
}
}
fn write_filled_qty(order: &mut OrderAny, filled_qty: Quantity) {
match order {
OrderAny::Limit(o) => o.filled_qty = filled_qty,
OrderAny::LimitIfTouched(o) => o.filled_qty = filled_qty,
OrderAny::Market(o) => o.filled_qty = filled_qty,
OrderAny::MarketIfTouched(o) => o.filled_qty = filled_qty,
OrderAny::MarketToLimit(o) => o.filled_qty = filled_qty,
OrderAny::StopLimit(o) => o.filled_qty = filled_qty,
OrderAny::StopMarket(o) => o.filled_qty = filled_qty,
OrderAny::TrailingStopLimit(o) => o.filled_qty = filled_qty,
OrderAny::TrailingStopMarket(o) => o.filled_qty = filled_qty,
}
}
#[derive(Debug, Clone, Copy)]
struct BarTickSizes {
open: Quantity,
high: Quantity,
low: Quantity,
close: Quantity,
}
impl BarTickSizes {
fn from_volume(volume: Quantity, size_increment: Quantity) -> Self {
let precision_diff = FIXED_PRECISION.saturating_sub(volume.precision);
let scale = QuantityRaw::pow(10, u32::from(precision_diff));
let units = volume.raw() / scale;
let increment_units = (size_increment.raw() / scale).max(1);
let rounded_units = (units / increment_units) * increment_units;
let increments = rounded_units / increment_units;
let zero = Quantity::zero(volume.precision);
let size =
|increments| Quantity::from_raw(increments * increment_units * scale, volume.precision);
match increments {
0 => Self {
open: zero,
high: zero,
low: zero,
close: zero,
},
1 => Self {
open: zero,
high: zero,
low: zero,
close: size(1),
},
2 => Self {
open: zero,
high: size(1),
low: size(1),
close: zero,
},
3 => {
let path_size = size(1);
Self {
open: path_size,
high: path_size,
low: path_size,
close: zero,
}
}
_ => {
let path_increments = increments / 4;
let close_increments = increments - (path_increments * 3);
let path_size = size(path_increments);
Self {
open: path_size,
high: path_size,
low: path_size,
close: size(close_increments),
}
}
}
}
}