use nautilus_common::enums::LogColor;
use nautilus_core::{UUID4, UnixNanos};
use nautilus_model::{
enums::{LiquiditySide, OrderStatus, OrderType},
events::{
OrderAccepted, OrderCanceled, OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled,
OrderRejected, OrderTriggered, OrderUpdated,
},
identifiers::{AccountId, PositionId, TradeId},
instruments::{Instrument, InstrumentAny},
orders::{Order, OrderAny, TRIGGERABLE_ORDER_TYPES},
reports::{FillReport, OrderStatusReport},
types::{Money, Price, Quantity},
};
use rust_decimal::Decimal;
use ustr::Ustr;
use super::{
ids::create_inferred_reconciliation_trade_id,
positions::{cap_price_at_instrument_max, is_within_single_unit_tolerance},
};
#[must_use]
pub fn generate_reconciliation_order_events(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
generate_reconciliation_order_events_inner(
order,
report,
instrument,
ts_now,
report.order_status == OrderStatus::Voided,
)
}
#[must_use]
pub fn generate_reconciliation_order_snapshot_events(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
generate_reconciliation_order_events_inner(order, report, instrument, ts_now, true)
}
fn generate_reconciliation_order_events_inner(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
allow_fill_decrease: bool,
) -> Vec<OrderEventAny> {
if is_superseded_cancel_report(order, report) {
let _ = reconcile_order_report(order, report, instrument, ts_now);
return Vec::new();
}
if has_material_fill_decrease(order, report) {
if !allow_fill_decrease {
log::warn!(
"Ignoring fill decrease without explicit void evidence for {}: cached={}, venue={}",
order.client_order_id(),
order.filled_qty(),
report.filled_qty,
);
return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
}
let mut working = order.clone();
let mut events = create_reconciliation_fill_voids(&working, report, ts_now);
if events.is_empty() {
return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
}
for event in &events {
if let Err(e) = working.apply(event.clone()) {
log::warn!(
"Cannot project reconciliation fill void for {}: {e}",
order.client_order_id()
);
return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
}
}
if report_is_working(report) && should_reconciliation_update(&working, report) {
let updated = create_reconciliation_updated(&working, report, ts_now);
if let Err(e) = working.apply(updated.clone()) {
log::warn!(
"Cannot project reopening reconciliation update for {}: {e}",
order.client_order_id()
);
} else {
events.push(updated);
}
}
if let Some(terminal) = reconcile_order_report(&working, report, instrument, ts_now) {
events.push(terminal);
}
return events;
}
let (mut working, mut events) = prepare_reconciliation_order(order, report, ts_now);
if matches!(
report.order_status,
OrderStatus::Canceled | OrderStatus::Expired,
) && report.filled_qty > working.filled_qty()
&& let Some(instrument) = instrument
&& let Some(filled) = create_incremental_inferred_fill(
&working,
report,
&report.account_id,
instrument,
ts_now,
None,
)
{
if let Err(e) = working.apply(filled.clone()) {
log::warn!(
"Failed to pre-apply reconciliation fill for {}: {e}",
order.client_order_id(),
);
} else {
events.push(filled);
}
}
if working.status() == OrderStatus::Filled
&& matches!(
report.order_status,
OrderStatus::Canceled | OrderStatus::Expired,
)
{
return events;
}
if let Some(event) = reconcile_order_report(&working, report, instrument, ts_now) {
events.push(event);
}
events
}
fn has_material_fill_decrease(order: &OrderAny, report: &OrderStatusReport) -> bool {
if report.filled_qty >= order.filled_qty() {
return false;
}
let precision = order
.filled_qty()
.precision
.max(report.filled_qty.precision);
!is_within_single_unit_tolerance(
report.filled_qty.as_decimal(),
order.filled_qty().as_decimal(),
precision,
)
}
fn reconcile_fill_decrease_terminal(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
let should_reconcile = report.order_status == OrderStatus::Voided
|| (!order.is_closed()
&& matches!(
report.order_status,
OrderStatus::Canceled | OrderStatus::Expired
));
if !should_reconcile {
return Vec::new();
}
reconcile_order_report(order, report, instrument, ts_now)
.into_iter()
.collect()
}
#[must_use]
pub fn generate_reconciliation_order_pre_fill_events(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
if is_superseded_cancel_report(order, report) {
return Vec::new();
}
let (working, mut events) = prepare_reconciliation_order(order, report, ts_now);
if report.order_status == OrderStatus::Triggered
&& let Some(triggered) = reconcile_order_report(&working, report, None, ts_now)
{
events.push(triggered);
}
events
}
fn prepare_reconciliation_order(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> (OrderAny, Vec<OrderEventAny>) {
let mut working = order.clone();
let mut events: Vec<OrderEventAny> = Vec::new();
if should_accept_before_reconciliation(&working, report) {
let Some(accepted) = create_reconciliation_accepted(&working, report, ts_now) else {
log::warn!(
"Cannot create reconciliation acceptance for {}: missing account_id",
order.client_order_id(),
);
return (working, events);
};
if let Err(e) = working.apply(accepted.clone()) {
log::warn!(
"Failed to pre-apply reconciliation acceptance for {}: {e}",
order.client_order_id(),
);
return (working, events);
}
events.push(accepted);
}
if report_is_confirmed_state(report)
&& (local_accepts_amendment(&working)
|| (matches!(
working.status(),
OrderStatus::PendingUpdate | OrderStatus::PendingCancel,
) && matches!(
report.order_status,
OrderStatus::Canceled | OrderStatus::Expired,
)))
&& should_reconciliation_update(&working, report)
{
let updated = create_reconciliation_updated(&working, report, ts_now);
if let Err(e) = working.apply(updated.clone()) {
log::warn!(
"Failed to pre-apply reconciliation update for {}: {e}",
order.client_order_id(),
);
} else {
events.push(updated);
}
}
(working, events)
}
#[must_use]
pub fn reconcile_order_report(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Option<OrderEventAny> {
if matches!(
report.order_status,
OrderStatus::PendingUpdate | OrderStatus::PendingCancel
) {
log::debug!(
"Order {} venue report in pending state: {:?}",
order.client_order_id(),
report.order_status,
);
return None;
}
if order.status() == report.order_status && order.filled_qty() == report.filled_qty {
if should_reconciliation_update(order, report) {
log::info!(
"Order {} has been updated at venue: qty={}->{}, price={:?}->{:?}",
order.client_order_id(),
order.quantity(),
report.quantity,
order.price(),
report.price
);
return Some(create_reconciliation_updated(order, report, ts_now));
}
return None; }
match report.order_status {
OrderStatus::Accepted => {
if order.status() == OrderStatus::Accepted
&& should_reconciliation_update(order, report)
{
return Some(create_reconciliation_updated(order, report, ts_now));
}
create_reconciliation_accepted(order, report, ts_now)
}
OrderStatus::Rejected => {
create_reconciliation_rejected(order, report.cancel_reason.as_deref(), ts_now)
}
OrderStatus::Triggered => {
if TRIGGERABLE_ORDER_TYPES.contains(&order.order_type()) {
Some(create_reconciliation_triggered(order, report, ts_now))
} else {
log::debug!(
"Skipping OrderTriggered for {} order {}: market-style stops have no TRIGGERED state",
order.order_type(),
order.client_order_id(),
);
None
}
}
OrderStatus::Canceled => {
if is_superseded_cancel_report(order, report) {
let cached_venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
log::info!(
"Suppressing Canceled for {} on previously-promoted venue_order_id {}: \
current venue_order_id is {}",
order.client_order_id(),
report.venue_order_id,
cached_venue_order_id,
);
return None;
}
Some(create_reconciliation_canceled(order, report, ts_now))
}
OrderStatus::Expired => Some(create_reconciliation_expired(order, report, ts_now)),
OrderStatus::PartiallyFilled | OrderStatus::Filled => {
reconcile_fill_quantity_mismatch(order, report, instrument, ts_now)
}
OrderStatus::Voided => {
create_reconciliation_terminal_fill_void(order, report, instrument, ts_now)
}
OrderStatus::PendingUpdate | OrderStatus::PendingCancel => None,
OrderStatus::Initialized
| OrderStatus::Submitted
| OrderStatus::Denied
| OrderStatus::Emulated
| OrderStatus::Released => {
log::warn!(
"Unexpected order status in venue report for {}: {:?}",
order.client_order_id(),
report.order_status
);
None
}
}
}
pub fn generate_external_order_status_events(
order: &OrderAny,
report: &OrderStatusReport,
account_id: &AccountId,
instrument: &InstrumentAny,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
let accepted = OrderEventAny::Accepted(OrderAccepted::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
report.venue_order_id,
*account_id,
UUID4::new(),
report.ts_accepted,
ts_now,
true, ));
match report.order_status {
OrderStatus::Accepted | OrderStatus::Triggered => vec![accepted],
OrderStatus::PartiallyFilled | OrderStatus::Filled => {
let mut events = vec![accepted];
if !report.filled_qty.is_zero()
&& let Some(filled) =
create_inferred_fill(order, report, *account_id, instrument, ts_now, None)
{
events.push(filled);
}
events
}
OrderStatus::Voided => {
let mut working = order.clone();
if let Err(e) = working.apply(accepted.clone()) {
log::warn!(
"Cannot project external order acceptance for {}: {e}",
order.client_order_id()
);
return vec![accepted];
}
let mut events = vec![accepted];
if !report.filled_qty.is_zero()
&& let Some(filled) =
create_inferred_fill(order, report, *account_id, instrument, ts_now, None)
{
if let Err(e) = working.apply(filled.clone()) {
log::warn!(
"Cannot project external order fill for {}: {e}",
order.client_order_id()
);
} else {
events.push(filled);
}
}
if let Some(voided) =
create_reconciliation_terminal_fill_void(&working, report, Some(instrument), ts_now)
{
events.push(voided);
}
events
}
OrderStatus::Canceled | OrderStatus::Expired => {
let terminal = create_external_terminal_event(order, report, *account_id, ts_now);
let mut events = vec![accepted];
let inferred_fill = if report.filled_qty.is_zero() {
None
} else {
create_inferred_fill(order, report, *account_id, instrument, ts_now, None)
};
let filled_to_quantity =
inferred_fill.is_some() && report.filled_qty >= report.quantity;
if let Some(filled) = inferred_fill {
events.push(filled);
}
if !filled_to_quantity {
events.push(terminal);
}
events
}
OrderStatus::Rejected => {
let reason = report.cancel_reason.as_deref().unwrap_or("UNKNOWN");
vec![OrderEventAny::Rejected(OrderRejected::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
*account_id,
Ustr::from(reason),
UUID4::new(),
report.ts_last,
ts_now,
true, reason_indicates_post_only_rejection(reason),
))]
}
_ => {
log::warn!(
"Unhandled order status {} for external order {}",
report.order_status,
order.client_order_id()
);
Vec::new()
}
}
}
fn create_reconciliation_fill_voids(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> Vec<OrderEventAny> {
let mut remaining = order.filled_qty() - report.filled_qty;
let order_events = order.events();
let mut corrections = Vec::new();
for candidate in order_events.iter().rev() {
let OrderEventAny::Filled(fill) = candidate else {
continue;
};
if remaining.is_zero() {
break;
}
let previous = order_events.iter().rev().find_map(|event| match event {
OrderEventAny::FillVoided(voided) if voided.trade_id == fill.trade_id => Some(voided),
_ => None,
});
let prior_qty = previous.map_or_else(
|| Quantity::zero(fill.last_qty.precision),
|voided| voided.voided_qty.min(fill.last_qty),
);
let effective = fill.last_qty - prior_qty;
if effective.is_zero() {
continue;
}
let removed = remaining.min(effective);
let voided_qty = prior_qty + removed;
let commission_voided = fill.commission.and_then(|commission| {
let fraction = voided_qty.as_decimal() / fill.last_qty.as_decimal();
Money::from_decimal(commission.as_decimal() * fraction, commission.currency).ok()
});
let mut event = OrderFillVoided::new(
fill.trader_id,
fill.strategy_id,
fill.instrument_id,
fill.client_order_id,
fill.venue_order_id,
fill.account_id,
Ustr::from(&format!(
"reconciliation-{}-{}",
report.report_id, fill.trade_id
)),
fill.trade_id,
voided_qty,
commission_voided,
fill.order_side,
fill.order_type,
fill.last_px,
fill.currency,
fill.liquidity_side,
fill.position_id,
report.cancel_reason.as_deref().map(Ustr::from),
None,
UUID4::new(),
report.ts_last,
ts_now,
true,
report_is_working(report),
);
event.causation_id = Some(report.report_id);
corrections.push(OrderEventAny::FillVoided(event));
remaining = remaining - removed;
}
if !remaining.is_zero() {
log::warn!(
"Cannot reconcile fill decrease for {}: {} is outside retained fill history",
order.client_order_id(),
remaining
);
return Vec::new();
}
corrections
}
fn create_reconciliation_terminal_fill_void(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Option<OrderEventAny> {
let voided_qty = order.leaves_qty();
if voided_qty.is_zero() {
return None;
}
let instrument = instrument?;
let last_px = resolve_fill_price(order, report, instrument)
.unwrap_or_else(|| Price::zero(instrument.price_precision()));
let mut event = OrderFillVoided::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
report.venue_order_id,
report.account_id,
Ustr::from(&format!("reconciliation-{}", report.report_id)),
TradeId::new(format!("VOID-{}", report.venue_order_id)),
voided_qty,
None,
order.order_side(),
order.order_type(),
last_px,
instrument.quote_currency(),
LiquiditySide::NoLiquiditySide,
None,
report.cancel_reason.as_deref().map(Ustr::from),
None,
UUID4::new(),
report.ts_last,
ts_now,
true,
false,
);
event.causation_id = Some(report.report_id);
Some(OrderEventAny::FillVoided(event))
}
fn create_external_terminal_event(
order: &OrderAny,
report: &OrderStatusReport,
account_id: AccountId,
ts_now: UnixNanos,
) -> OrderEventAny {
match report.order_status {
OrderStatus::Canceled => OrderEventAny::Canceled(OrderCanceled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
report.ts_last,
ts_now,
true, Some(report.venue_order_id),
Some(account_id),
)),
OrderStatus::Expired => OrderEventAny::Expired(OrderExpired::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
report.ts_last,
ts_now,
true, Some(report.venue_order_id),
Some(account_id),
)),
status => unreachable!("cannot create external terminal event for {status}"),
}
}
pub fn reconcile_fill_report(
order: &OrderAny,
report: &FillReport,
instrument: &InstrumentAny,
ts_now: UnixNanos,
allow_overfills: bool,
) -> Option<OrderEventAny> {
debug_assert!(
!report.last_qty.is_zero(),
"fill report last_qty must be non-zero for {}",
order.client_order_id(),
);
if order.trade_ids().iter().any(|id| **id == report.trade_id) {
log::debug!(
"Duplicate fill detected: trade_id {} already exists for order {}",
report.trade_id,
order.client_order_id()
);
return None;
}
let potential_filled_qty = order.filled_qty() + report.last_qty;
if potential_filled_qty > order.quantity() {
if !allow_overfills {
log::warn!(
"Rejecting fill that would cause overfill for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, would result in filled_qty={}",
order.client_order_id(),
order.quantity(),
order.filled_qty(),
report.last_qty,
potential_filled_qty
);
return None;
}
log::warn!(
"Allowing overfill during reconciliation for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, will result in filled_qty={}",
order.client_order_id(),
order.quantity(),
order.filled_qty(),
report.last_qty,
potential_filled_qty
);
}
let account_id = report.account_id;
let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
log::info!(
color = LogColor::Blue as u8;
"Reconciling fill for {}: qty={}, px={}, trade_id={}",
order.client_order_id(),
report.last_qty,
report.last_px,
report.trade_id,
);
Some(OrderEventAny::Filled(OrderFilled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
venue_order_id,
account_id,
report.trade_id,
order.order_side(),
order.order_type(),
report.last_qty,
report.last_px,
instrument.quote_currency(),
report.liquidity_side,
UUID4::new(),
report.ts_event,
ts_now,
true, report.venue_position_id,
Some(report.commission),
None,
)))
}
pub fn should_reconciliation_update(order: &OrderAny, report: &OrderStatusReport) -> bool {
if report.quantity != order.quantity() && report.quantity >= order.filled_qty() {
return true;
}
let price_drift = report.price.is_some() && report.price != order.price();
let trigger_drift =
report.trigger_price.is_some() && report.trigger_price != order.trigger_price();
match order.order_type() {
OrderType::Limit => price_drift,
OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched => {
trigger_drift
}
OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched => {
trigger_drift || price_drift
}
_ => false,
}
}
#[must_use]
pub(super) fn create_reconciliation_accepted(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> Option<OrderEventAny> {
let account_id = order.account_id()?;
Some(OrderEventAny::Accepted(OrderAccepted::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
order.venue_order_id().unwrap_or(report.venue_order_id),
account_id,
UUID4::new(),
report.ts_accepted,
ts_now,
true, )))
}
#[must_use]
pub fn create_reconciliation_rejected(
order: &OrderAny,
reason: Option<&str>,
ts_now: UnixNanos,
) -> Option<OrderEventAny> {
let account_id = order.account_id()?;
let reason = reason.unwrap_or("UNKNOWN");
Some(OrderEventAny::Rejected(OrderRejected::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
account_id,
Ustr::from(reason),
UUID4::new(),
ts_now,
ts_now,
true, reason_indicates_post_only_rejection(reason),
)))
}
fn reason_indicates_post_only_rejection(reason: &str) -> bool {
let normalized: String = reason
.chars()
.filter_map(|ch| {
if ch == '-' || ch == '_' || ch.is_whitespace() {
None
} else {
Some(ch.to_ascii_lowercase())
}
})
.collect();
normalized.contains("postonly") || normalized.contains("postwouldexecute")
}
#[must_use]
pub fn create_reconciliation_triggered(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> OrderEventAny {
OrderEventAny::Triggered(OrderTriggered::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
report.ts_triggered.unwrap_or(ts_now),
ts_now,
true, order.venue_order_id(),
order.account_id(),
))
}
#[must_use]
pub(super) fn create_reconciliation_canceled(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> OrderEventAny {
OrderEventAny::Canceled(OrderCanceled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
report.ts_last,
ts_now,
true, order.venue_order_id(),
order.account_id(),
))
}
#[must_use]
pub(super) fn create_reconciliation_expired(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> OrderEventAny {
OrderEventAny::Expired(OrderExpired::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
UUID4::new(),
report.ts_last,
ts_now,
true, order.venue_order_id(),
order.account_id(),
))
}
#[must_use]
pub(super) fn create_reconciliation_updated(
order: &OrderAny,
report: &OrderStatusReport,
ts_now: UnixNanos,
) -> OrderEventAny {
let trigger_price = match order.order_type() {
OrderType::StopMarket
| OrderType::StopLimit
| OrderType::MarketIfTouched
| OrderType::LimitIfTouched
| OrderType::TrailingStopMarket
| OrderType::TrailingStopLimit => report.trigger_price,
_ => None,
};
OrderEventAny::Updated(OrderUpdated::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
report.quantity,
UUID4::new(),
report.ts_last,
ts_now,
true, order.venue_order_id(),
order.account_id(),
report.price,
trigger_price,
None, order.is_quote_quantity(),
))
}
pub(super) fn create_inferred_fill(
order: &OrderAny,
report: &OrderStatusReport,
account_id: AccountId,
instrument: &InstrumentAny,
ts_now: UnixNanos,
commission: Option<Money>,
) -> Option<OrderEventAny> {
let liquidity_side = match order.order_type() {
OrderType::Market
| OrderType::StopMarket
| OrderType::MarketToLimit
| OrderType::TrailingStopMarket => LiquiditySide::Taker,
_ if order.is_post_only() => LiquiditySide::Maker,
_ => LiquiditySide::NoLiquiditySide,
};
let Some(last_px) = resolve_fill_price(order, report, instrument) else {
log::warn!(
"Cannot create inferred fill for {}: no avg_px, report price, or order price",
order.client_order_id()
);
return None;
};
let last_px = clamp_inferred_fill_price(last_px, instrument);
let position_id = reconciliation_position_id(report, instrument);
let trade_id = create_inferred_reconciliation_trade_id(
account_id,
order.instrument_id(),
order.client_order_id(),
Some(report.venue_order_id),
report.order_side,
order.order_type(),
report.filled_qty,
report.filled_qty,
last_px,
position_id,
report.ts_last,
);
log::info!(
"Generated inferred fill for {} ({}) qty={} px={}",
order.client_order_id(),
report.venue_order_id,
report.filled_qty,
last_px,
);
Some(OrderEventAny::Filled(OrderFilled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
report.venue_order_id,
account_id,
trade_id,
report.order_side,
order.order_type(),
report.filled_qty,
last_px,
instrument.quote_currency(),
liquidity_side,
UUID4::new(),
report.ts_last,
ts_now,
true, report.venue_position_id,
commission,
None,
)))
}
pub fn create_incremental_inferred_fill(
order: &OrderAny,
report: &OrderStatusReport,
account_id: &AccountId,
instrument: &InstrumentAny,
ts_now: UnixNanos,
commission: Option<Money>,
) -> Option<OrderEventAny> {
let order_filled_qty = order.filled_qty();
debug_assert!(
report.filled_qty >= order_filled_qty,
"incremental inferred fill requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
report.filled_qty,
order_filled_qty,
order.client_order_id(),
);
let last_qty = report.filled_qty - order_filled_qty;
if last_qty <= Quantity::zero(instrument.size_precision()) {
return None;
}
let liquidity_side = match order.order_type() {
OrderType::Market
| OrderType::StopMarket
| OrderType::MarketToLimit
| OrderType::TrailingStopMarket => LiquiditySide::Taker,
_ if order.is_post_only() => LiquiditySide::Maker,
_ => LiquiditySide::NoLiquiditySide,
};
let last_px = calculate_incremental_fill_price(order, report, instrument)?;
let last_px = clamp_inferred_fill_price(last_px, instrument);
let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
let position_id = reconciliation_position_id(report, instrument);
let trade_id = create_inferred_reconciliation_trade_id(
*account_id,
order.instrument_id(),
order.client_order_id(),
Some(venue_order_id),
order.order_side(),
order.order_type(),
report.filled_qty,
last_qty,
last_px,
position_id,
report.ts_last,
);
log::info!(
color = LogColor::Blue as u8;
"Generated inferred fill for {}: qty={}, px={}",
order.client_order_id(),
last_qty,
last_px,
);
Some(OrderEventAny::Filled(OrderFilled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
venue_order_id,
*account_id,
trade_id,
order.order_side(),
order.order_type(),
last_qty,
last_px,
instrument.quote_currency(),
liquidity_side,
UUID4::new(),
report.ts_last,
ts_now,
true, None, commission,
None,
)))
}
pub fn create_inferred_fill_for_qty(
order: &OrderAny,
report: &OrderStatusReport,
account_id: &AccountId,
instrument: &InstrumentAny,
fill_qty: Quantity,
ts_now: UnixNanos,
commission: Option<Money>,
) -> Option<OrderEventAny> {
if fill_qty.is_zero() {
return None;
}
let liquidity_side = match order.order_type() {
OrderType::Market
| OrderType::StopMarket
| OrderType::MarketToLimit
| OrderType::TrailingStopMarket => LiquiditySide::Taker,
_ if order.is_post_only() => LiquiditySide::Maker,
_ => LiquiditySide::NoLiquiditySide,
};
let Some(last_px) = resolve_fill_price(order, report, instrument) else {
log::warn!(
"Cannot determine fill price for {}: no avg_px, report price, or order price",
order.client_order_id()
);
return None;
};
let last_px = clamp_inferred_fill_price(last_px, instrument);
let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
let position_id = reconciliation_position_id(report, instrument);
let trade_id = create_inferred_reconciliation_trade_id(
*account_id,
order.instrument_id(),
order.client_order_id(),
Some(venue_order_id),
order.order_side(),
order.order_type(),
report.filled_qty,
fill_qty,
last_px,
position_id,
report.ts_last,
);
log::info!(
color = LogColor::Blue as u8;
"Generated inferred fill for {}: qty={}, px={}",
order.client_order_id(),
fill_qty,
last_px,
);
Some(OrderEventAny::Filled(OrderFilled::new(
order.trader_id(),
order.strategy_id(),
order.instrument_id(),
order.client_order_id(),
venue_order_id,
*account_id,
trade_id,
order.order_side(),
order.order_type(),
fill_qty,
last_px,
instrument.quote_currency(),
liquidity_side,
UUID4::new(),
report.ts_last,
ts_now,
true, None, commission,
None,
)))
}
fn report_is_confirmed_state(report: &OrderStatusReport) -> bool {
matches!(
report.order_status,
OrderStatus::Accepted
| OrderStatus::Triggered
| OrderStatus::PartiallyFilled
| OrderStatus::Filled
| OrderStatus::Canceled
| OrderStatus::Expired
)
}
fn report_is_working(report: &OrderStatusReport) -> bool {
matches!(
report.order_status,
OrderStatus::Accepted | OrderStatus::Triggered | OrderStatus::PartiallyFilled
)
}
pub(crate) fn is_superseded_cancel_report(order: &OrderAny, report: &OrderStatusReport) -> bool {
if report.order_status != OrderStatus::Canceled {
return false;
}
let Some(cached_venue_order_id) = order.venue_order_id() else {
return false;
};
cached_venue_order_id != report.venue_order_id
&& order
.venue_order_ids()
.iter()
.any(|venue_order_id| **venue_order_id == report.venue_order_id)
}
fn local_accepts_amendment(order: &OrderAny) -> bool {
matches!(
order.status(),
OrderStatus::Accepted | OrderStatus::Triggered | OrderStatus::PartiallyFilled
)
}
fn should_accept_before_reconciliation(order: &OrderAny, report: &OrderStatusReport) -> bool {
order.status() == OrderStatus::Submitted && report.order_status != OrderStatus::Rejected
}
fn reconcile_fill_quantity_mismatch(
order: &OrderAny,
report: &OrderStatusReport,
instrument: Option<&InstrumentAny>,
ts_now: UnixNanos,
) -> Option<OrderEventAny> {
let order_filled_qty = order.filled_qty();
let report_filled_qty = report.filled_qty;
if report_filled_qty < order_filled_qty {
let precision = order_filled_qty.precision.max(report_filled_qty.precision);
if is_within_single_unit_tolerance(
report_filled_qty.as_decimal(),
order_filled_qty.as_decimal(),
precision,
) {
return None;
}
log::warn!(
"Fill qty mismatch for {} ({}): cached={}, venue={}, order_qty={} (venue < cached)",
order.client_order_id(),
report.venue_order_id,
order_filled_qty,
report_filled_qty,
order.quantity(),
);
return None;
}
if report_filled_qty > order_filled_qty {
if order.is_closed() {
let precision = order_filled_qty.precision.max(report_filled_qty.precision);
if is_within_single_unit_tolerance(
report_filled_qty.as_decimal(),
order_filled_qty.as_decimal(),
precision,
) {
return None;
}
log::debug!(
"{} {} already closed but reported difference in filled_qty: \
report={}, cached={}, skipping inferred fill generation for closed order",
order.instrument_id(),
order.client_order_id(),
report_filled_qty,
order_filled_qty,
);
return None;
}
let Some(instrument) = instrument else {
log::warn!(
"Cannot generate inferred fill for {}: instrument not available",
order.client_order_id()
);
return None;
};
let account_id = order.account_id()?;
return create_incremental_inferred_fill(
order,
report,
&account_id,
instrument,
ts_now,
None,
);
}
if order.status() != report.order_status {
if should_reconciliation_update(order, report) {
log::info!(
"Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, \
filled_qty={}, updating quantity {}->{}",
order.client_order_id(),
order.status(),
report.order_status,
report.filled_qty,
order.quantity(),
report.quantity,
);
return Some(create_reconciliation_updated(order, report, ts_now));
}
log::warn!(
"Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, filled_qty={}",
order.client_order_id(),
order.status(),
report.order_status,
report.filled_qty
);
}
None
}
fn calculate_incremental_fill_price(
order: &OrderAny,
report: &OrderStatusReport,
instrument: &InstrumentAny,
) -> Option<Price> {
let order_filled_qty = order.filled_qty();
debug_assert!(
report.filled_qty >= order_filled_qty,
"incremental fill price requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
report.filled_qty,
order_filled_qty,
order.client_order_id(),
);
if order_filled_qty.is_zero() {
let last_px = resolve_fill_price(order, report, instrument);
if last_px.is_none() {
log::warn!(
"Cannot determine fill price for {}: no avg_px, report price, or order price",
order.client_order_id()
);
}
return last_px;
}
if let Some(report_avg_px) = report.avg_px {
let last_px_decimal = match order.avg_px() {
None => report_avg_px,
Some(order_avg_px) => {
let report_filled_qty = report.filled_qty;
let last_qty = report_filled_qty - order_filled_qty;
let report_notional = report_avg_px * report_filled_qty.as_decimal();
let order_notional = order_avg_px * order_filled_qty.as_decimal();
let last_notional = report_notional - order_notional;
let back_solved = last_notional / last_qty.as_decimal();
if back_solved < Decimal::ZERO && !instrument.allows_negative_price() {
if report_avg_px < Decimal::ZERO {
log::warn!(
"Cannot price inferred fill for {}: back-solved {back_solved} and venue average {report_avg_px} are both negative on an instrument that disallows negative prices",
order.client_order_id(),
);
return None;
}
log::warn!(
"Negative back-solved fill price {back_solved} for {}, using venue average {report_avg_px}",
order.client_order_id(),
);
report_avg_px
} else {
back_solved
}
}
};
return Price::from_decimal_dp(last_px_decimal, instrument.price_precision())
.inspect_err(|e| {
log::warn!(
"Cannot price {} from incremental {last_px_decimal}, falling back: {e}",
order.client_order_id(),
);
})
.ok()
.or_else(|| resolve_fill_price(order, report, instrument));
}
resolve_fill_price(order, report, instrument)
}
fn resolve_fill_price(
order: &OrderAny,
report: &OrderStatusReport,
instrument: &InstrumentAny,
) -> Option<Price> {
report
.avg_px
.and_then(|avg_px| {
Price::from_decimal_dp(avg_px, instrument.price_precision())
.inspect_err(|e| {
log::warn!(
"Cannot price {} from venue average {avg_px}, trying next source: {e}",
order.client_order_id(),
);
})
.ok()
})
.or(report.price)
.or_else(|| order.price())
}
fn clamp_inferred_fill_price(price: Price, instrument: &InstrumentAny) -> Price {
let px = cap_price_at_instrument_max(price.as_decimal(), instrument);
Price::from_decimal_dp(px, instrument.price_precision()).unwrap_or(price)
}
fn reconciliation_position_id(
report: &OrderStatusReport,
instrument: &InstrumentAny,
) -> PositionId {
report
.venue_position_id
.unwrap_or_else(|| PositionId::new(format!("{}-EXTERNAL", instrument.id())))
}