1use std::collections::HashSet;
22
23use nautilus_common::enums::LogColor;
24use nautilus_core::{UUID4, UnixNanos};
25use nautilus_model::{
26 enums::{LiquiditySide, OrderStatus, OrderType},
27 events::{
28 OrderAccepted, OrderCanceled, OrderEventAny, OrderExpired, OrderFillVoided, OrderFilled,
29 OrderRejected, OrderTriggered, OrderUpdated,
30 },
31 identifiers::{AccountId, PositionId, TradeId},
32 instruments::{Instrument, InstrumentAny},
33 orders::{Order, OrderAny, TRIGGERABLE_ORDER_TYPES},
34 reports::{FillReport, OrderStatusReport},
35 types::{Money, Price, Quantity},
36};
37use rust_decimal::Decimal;
38use ustr::Ustr;
39
40use super::{
41 ids::create_inferred_reconciliation_trade_id,
42 positions::{cap_price_at_instrument_max, is_within_single_unit_tolerance},
43 reconciliation_price,
44};
45
46#[must_use]
59pub fn generate_reconciliation_order_events(
60 order: &OrderAny,
61 report: &OrderStatusReport,
62 instrument: Option<&InstrumentAny>,
63 ts_now: UnixNanos,
64) -> Vec<OrderEventAny> {
65 generate_reconciliation_order_events_inner(
66 order,
67 report,
68 instrument,
69 ts_now,
70 report.order_status == OrderStatus::Voided,
71 None,
72 )
73}
74
75#[must_use]
82pub fn generate_reconciliation_order_snapshot_events(
83 order: &OrderAny,
84 report: &OrderStatusReport,
85 instrument: Option<&InstrumentAny>,
86 ts_now: UnixNanos,
87) -> Vec<OrderEventAny> {
88 generate_reconciliation_order_events_inner(order, report, instrument, ts_now, true, None)
89}
90
91#[must_use]
94pub fn generate_reconciliation_order_snapshot_events_with_commission(
95 order: &OrderAny,
96 report: &OrderStatusReport,
97 instrument: Option<&InstrumentAny>,
98 ts_now: UnixNanos,
99 commission: Option<Money>,
100) -> Vec<OrderEventAny> {
101 generate_reconciliation_order_events_inner(order, report, instrument, ts_now, true, commission)
102}
103
104fn generate_reconciliation_order_events_inner(
105 order: &OrderAny,
106 report: &OrderStatusReport,
107 instrument: Option<&InstrumentAny>,
108 ts_now: UnixNanos,
109 allow_fill_decrease: bool,
110 commission: Option<Money>,
111) -> Vec<OrderEventAny> {
112 if is_superseded_terminal_report(order, report) {
113 let _ = reconcile_order_report(order, report, instrument, ts_now);
114 return Vec::new();
115 }
116
117 if has_material_fill_decrease(order, report) {
118 if !allow_fill_decrease {
119 log::warn!(
120 "Ignoring fill decrease without explicit void evidence for {}: cached={}, venue={}",
121 order.client_order_id(),
122 order.filled_qty(),
123 report.filled_qty,
124 );
125 return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
126 }
127
128 let mut working = order.clone();
129 let mut events = create_reconciliation_fill_voids(&working, report, ts_now);
130 if events.is_empty() {
131 return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
132 }
133
134 for event in &events {
135 if let Err(e) = working.apply(event.clone()) {
136 log::warn!(
137 "Cannot project reconciliation fill void for {}: {e}",
138 order.client_order_id()
139 );
140 return reconcile_fill_decrease_terminal(order, report, instrument, ts_now);
141 }
142 }
143
144 if report_is_working(report) && should_reconciliation_update(&working, report) {
145 let updated = create_reconciliation_updated(&working, report, ts_now);
146 if let Err(e) = working.apply(updated.clone()) {
147 log::warn!(
148 "Cannot project reopening reconciliation update for {}: {e}",
149 order.client_order_id()
150 );
151 } else {
152 events.push(updated);
153 }
154 }
155
156 if let Some(terminal) = reconcile_order_report(&working, report, instrument, ts_now) {
157 events.push(terminal);
158 }
159 return events;
160 }
161
162 let (mut working, mut events) = prepare_reconciliation_order(order, report, ts_now);
163
164 if matches!(
165 report.order_status,
166 OrderStatus::Canceled | OrderStatus::Expired,
167 ) && report.filled_qty > working.filled_qty()
168 && let Some(instrument) = instrument
169 && let Some(filled) = create_incremental_inferred_fill(
170 &working,
171 report,
172 &report.account_id,
173 instrument,
174 ts_now,
175 commission,
176 )
177 {
178 if let Err(e) = working.apply(filled.clone()) {
179 log::warn!(
180 "Failed to pre-apply reconciliation fill for {}: {e}",
181 order.client_order_id(),
182 );
183 } else {
184 events.push(filled);
185 }
186 }
187
188 if working.status() == OrderStatus::Filled
189 && matches!(
190 report.order_status,
191 OrderStatus::Canceled | OrderStatus::Expired,
192 )
193 {
194 return events;
195 }
196
197 if let Some(event) =
198 reconcile_order_report_with_commission(&working, report, instrument, ts_now, commission)
199 {
200 events.push(event);
201 }
202
203 events
204}
205
206fn has_material_fill_decrease(order: &OrderAny, report: &OrderStatusReport) -> bool {
207 if report.filled_qty >= order.filled_qty() {
208 return false;
209 }
210
211 let precision = order
212 .filled_qty()
213 .precision
214 .max(report.filled_qty.precision);
215 !is_within_single_unit_tolerance(
216 report.filled_qty.as_decimal(),
217 order.filled_qty().as_decimal(),
218 precision,
219 )
220}
221
222fn reconcile_fill_decrease_terminal(
223 order: &OrderAny,
224 report: &OrderStatusReport,
225 instrument: Option<&InstrumentAny>,
226 ts_now: UnixNanos,
227) -> Vec<OrderEventAny> {
228 let should_reconcile = report.order_status == OrderStatus::Voided
229 || (!order.is_closed()
230 && matches!(
231 report.order_status,
232 OrderStatus::Canceled | OrderStatus::Expired
233 ));
234
235 if !should_reconcile {
236 return Vec::new();
237 }
238
239 reconcile_order_report(order, report, instrument, ts_now)
240 .into_iter()
241 .collect()
242}
243
244#[must_use]
246pub fn generate_reconciliation_order_pre_fill_events(
247 order: &OrderAny,
248 report: &OrderStatusReport,
249 ts_now: UnixNanos,
250) -> Vec<OrderEventAny> {
251 if is_superseded_terminal_report(order, report) {
252 return Vec::new();
253 }
254
255 let (working, mut events) = prepare_reconciliation_order(order, report, ts_now);
256
257 if report.order_status == OrderStatus::Triggered
258 && let Some(triggered) = reconcile_order_report(&working, report, None, ts_now)
259 {
260 events.push(triggered);
261 }
262
263 events
264}
265
266fn prepare_reconciliation_order(
267 order: &OrderAny,
268 report: &OrderStatusReport,
269 ts_now: UnixNanos,
270) -> (OrderAny, Vec<OrderEventAny>) {
271 let mut working = order.clone();
272 let mut events: Vec<OrderEventAny> = Vec::new();
273
274 if should_accept_before_reconciliation(&working, report) {
275 let Some(accepted) = create_reconciliation_accepted(&working, report, ts_now) else {
276 log::warn!(
277 "Cannot create reconciliation acceptance for {}: missing account_id",
278 order.client_order_id(),
279 );
280 return (working, events);
281 };
282
283 if let Err(e) = working.apply(accepted.clone()) {
284 log::warn!(
285 "Failed to pre-apply reconciliation acceptance for {}: {e}",
286 order.client_order_id(),
287 );
288 return (working, events);
289 }
290 events.push(accepted);
291 }
292
293 if report_is_confirmed_state(report)
294 && (local_accepts_amendment(&working) || pending_terminal_quantity_update(&working, report))
295 && should_reconciliation_update(&working, report)
296 {
297 let updated = create_reconciliation_updated(&working, report, ts_now);
298 if let Err(e) = working.apply(updated.clone()) {
299 log::warn!(
300 "Failed to pre-apply reconciliation update for {}: {e}",
301 order.client_order_id(),
302 );
303 } else {
304 events.push(updated);
305 }
306 }
307
308 (working, events)
309}
310
311#[must_use]
325pub fn reconcile_order_report(
326 order: &OrderAny,
327 report: &OrderStatusReport,
328 instrument: Option<&InstrumentAny>,
329 ts_now: UnixNanos,
330) -> Option<OrderEventAny> {
331 reconcile_order_report_with_commission(order, report, instrument, ts_now, None)
332}
333
334#[must_use]
336pub fn reconcile_order_report_with_commission(
337 order: &OrderAny,
338 report: &OrderStatusReport,
339 instrument: Option<&InstrumentAny>,
340 ts_now: UnixNanos,
341 commission: Option<Money>,
342) -> Option<OrderEventAny> {
343 if matches!(
344 report.order_status,
345 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
346 ) {
347 log::debug!(
348 "Order {} venue report in pending state: {:?}",
349 order.client_order_id(),
350 report.order_status,
351 );
352 return None;
353 }
354
355 if is_unchanged_accepted_report_during_pending_command(order, report) {
356 log::debug!(
357 "Order {} remains inflight while venue reports an unchanged accepted snapshot",
358 order.client_order_id(),
359 );
360 return None;
361 }
362
363 if is_superseded_terminal_report(order, report) {
364 let cached_venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
365 log::info!(
366 "Suppressing {:?} for {} on previously-promoted venue_order_id {}: \
367 current venue_order_id is {}",
368 report.order_status,
369 order.client_order_id(),
370 report.venue_order_id,
371 cached_venue_order_id,
372 );
373 return None;
374 }
375
376 if order.status() == report.order_status && order.filled_qty() == report.filled_qty {
377 if should_reconciliation_update(order, report) {
378 log::info!(
379 "Order {} has been updated at venue: qty={}->{}, price={:?}->{:?}",
380 order.client_order_id(),
381 order.quantity(),
382 report.quantity,
383 order.price(),
384 report.price
385 );
386 return Some(create_reconciliation_updated(order, report, ts_now));
387 }
388 return None; }
390
391 match report.order_status {
392 OrderStatus::Accepted => {
393 if order.status() == OrderStatus::Accepted
394 && should_reconciliation_update(order, report)
395 {
396 return Some(create_reconciliation_updated(order, report, ts_now));
397 }
398 create_reconciliation_accepted(order, report, ts_now)
399 }
400 OrderStatus::Rejected => {
401 create_reconciliation_rejected(order, report.cancel_reason.as_deref(), ts_now)
402 }
403 OrderStatus::Triggered => {
404 if TRIGGERABLE_ORDER_TYPES.contains(&order.order_type()) {
405 Some(create_reconciliation_triggered(order, report, ts_now))
406 } else {
407 log::debug!(
408 "Skipping OrderTriggered for {} order {}: market-style stops have no TRIGGERED state",
409 order.order_type(),
410 order.client_order_id(),
411 );
412 None
413 }
414 }
415 OrderStatus::Canceled => Some(create_reconciliation_canceled(order, report, ts_now)),
416 OrderStatus::Expired => Some(create_reconciliation_expired(order, report, ts_now)),
417
418 OrderStatus::PartiallyFilled | OrderStatus::Filled => {
419 reconcile_fill_quantity_mismatch(order, report, instrument, ts_now, commission)
420 }
421
422 OrderStatus::Voided => {
423 create_reconciliation_terminal_fill_void(order, report, instrument, ts_now)
424 }
425
426 OrderStatus::PendingUpdate | OrderStatus::PendingCancel => None,
427
428 OrderStatus::Initialized
430 | OrderStatus::Submitted
431 | OrderStatus::Denied
432 | OrderStatus::Emulated
433 | OrderStatus::Released => {
434 log::warn!(
435 "Unexpected order status in venue report for {}: {:?}",
436 order.client_order_id(),
437 report.order_status
438 );
439 None
440 }
441 }
442}
443
444fn is_unchanged_accepted_report_during_pending_command(
445 order: &OrderAny,
446 report: &OrderStatusReport,
447) -> bool {
448 matches!(
449 order.status(),
450 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
451 ) && report.order_status == OrderStatus::Accepted
452 && order.venue_order_id() == Some(report.venue_order_id)
453 && order.filled_qty() == report.filled_qty
454 && !should_reconciliation_update(order, report)
455}
456
457pub fn generate_external_order_status_events(
464 order: &OrderAny,
465 report: &OrderStatusReport,
466 account_id: &AccountId,
467 instrument: &InstrumentAny,
468 ts_now: UnixNanos,
469) -> Vec<OrderEventAny> {
470 generate_external_order_status_events_with_commission(
471 order, report, account_id, instrument, ts_now, None,
472 )
473}
474
475#[must_use]
477pub fn generate_external_order_status_events_with_commission(
478 order: &OrderAny,
479 report: &OrderStatusReport,
480 account_id: &AccountId,
481 instrument: &InstrumentAny,
482 ts_now: UnixNanos,
483 commission: Option<Money>,
484) -> Vec<OrderEventAny> {
485 let accepted = OrderEventAny::Accepted(OrderAccepted::new(
486 order.trader_id(),
487 order.strategy_id(),
488 order.instrument_id(),
489 order.client_order_id(),
490 report.venue_order_id,
491 *account_id,
492 UUID4::new(),
493 report.ts_accepted,
494 ts_now,
495 true, ));
497
498 match report.order_status {
499 OrderStatus::Accepted | OrderStatus::Triggered => vec![accepted],
500 OrderStatus::PartiallyFilled | OrderStatus::Filled => {
501 let mut events = vec![accepted];
502
503 if !report.filled_qty.is_zero()
504 && let Some(filled) =
505 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
506 {
507 events.push(filled);
508 }
509
510 events
511 }
512 OrderStatus::Voided => {
513 let mut working = order.clone();
514 if let Err(e) = working.apply(accepted.clone()) {
515 log::warn!(
516 "Cannot project external order acceptance for {}: {e}",
517 order.client_order_id()
518 );
519 return vec![accepted];
520 }
521 let mut events = vec![accepted];
522
523 if !report.filled_qty.is_zero()
524 && let Some(filled) =
525 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
526 {
527 if let Err(e) = working.apply(filled.clone()) {
528 log::warn!(
529 "Cannot project external order fill for {}: {e}",
530 order.client_order_id()
531 );
532 } else {
533 events.push(filled);
534 }
535 }
536
537 if let Some(voided) =
538 create_reconciliation_terminal_fill_void(&working, report, Some(instrument), ts_now)
539 {
540 events.push(voided);
541 }
542 events
543 }
544 OrderStatus::Canceled | OrderStatus::Expired => {
545 let terminal = create_external_terminal_event(order, report, *account_id, ts_now);
546 let mut events = vec![accepted];
547
548 let inferred_fill = if report.filled_qty.is_zero() {
549 None
550 } else {
551 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
552 };
553 let filled_to_quantity =
554 inferred_fill.is_some() && report.filled_qty >= report.quantity;
555 if let Some(filled) = inferred_fill {
556 events.push(filled);
557 }
558
559 if !filled_to_quantity {
560 events.push(terminal);
561 }
562 events
563 }
564 OrderStatus::Rejected => {
565 let reason = report.cancel_reason.as_deref().unwrap_or("UNKNOWN");
567 vec![OrderEventAny::Rejected(OrderRejected::new(
568 order.trader_id(),
569 order.strategy_id(),
570 order.instrument_id(),
571 order.client_order_id(),
572 *account_id,
573 Ustr::from(reason),
574 UUID4::new(),
575 report.ts_last,
576 ts_now,
577 true, reason_indicates_post_only_rejection(reason),
579 ))]
580 }
581 _ => {
582 log::warn!(
583 "Unhandled order status {} for external order {}",
584 report.order_status,
585 order.client_order_id()
586 );
587 Vec::new()
588 }
589 }
590}
591
592fn create_reconciliation_fill_voids(
593 order: &OrderAny,
594 report: &OrderStatusReport,
595 ts_now: UnixNanos,
596) -> Vec<OrderEventAny> {
597 let mut remaining = order.filled_qty() - report.filled_qty;
598 let order_events = order.events();
599 let mut corrections = Vec::new();
600
601 let mut seen = HashSet::new();
604
605 let quantity_fills: Vec<&OrderFilled> = order_events
606 .iter()
607 .filter_map(|event| match event {
608 OrderEventAny::Filled(fill) if seen.insert(fill.trade_id) => Some(fill),
609 _ => None,
610 })
611 .collect();
612
613 for fill in quantity_fills.iter().rev() {
614 if remaining.is_zero() {
615 break;
616 }
617 let previous = order_events.iter().rev().find_map(|event| match event {
618 OrderEventAny::FillVoided(voided) if voided.trade_id == fill.trade_id => Some(voided),
619 _ => None,
620 });
621 let prior_qty = previous.map_or_else(
622 || Quantity::zero(fill.last_qty.precision),
623 |voided| voided.voided_qty.min(fill.last_qty),
624 );
625 let effective = fill.last_qty - prior_qty;
626 if effective.is_zero() {
627 continue;
628 }
629 let removed = remaining.min(effective);
630 let voided_qty = prior_qty + removed;
631 let commission_voided = fill.commission.and_then(|commission| {
632 let fraction = voided_qty.as_decimal() / fill.last_qty.as_decimal();
633 Money::from_decimal(commission.as_decimal() * fraction, commission.currency).ok()
634 });
635 let mut event = OrderFillVoided::new(
636 fill.trader_id,
637 fill.strategy_id,
638 fill.instrument_id,
639 fill.client_order_id,
640 fill.venue_order_id,
641 fill.account_id,
642 Ustr::from(&format!(
643 "reconciliation-{}-{}",
644 report.report_id, fill.trade_id
645 )),
646 fill.trade_id,
647 voided_qty,
648 commission_voided,
649 fill.order_side,
650 fill.order_type,
651 fill.last_px,
652 fill.currency,
653 fill.liquidity_side,
654 fill.position_id,
655 report.cancel_reason.as_deref().map(Ustr::from),
656 None,
657 UUID4::new(),
658 report.ts_last,
659 ts_now,
660 true,
661 report_is_working(report),
662 );
663 event.causation_id = Some(report.report_id);
664 corrections.push(OrderEventAny::FillVoided(event));
665 remaining = remaining - removed;
666 }
667
668 if !remaining.is_zero() {
669 log::warn!(
670 "Cannot reconcile fill decrease for {}: {} is outside retained fill history",
671 order.client_order_id(),
672 remaining
673 );
674 return Vec::new();
675 }
676 corrections
677}
678
679fn create_reconciliation_terminal_fill_void(
685 order: &OrderAny,
686 report: &OrderStatusReport,
687 instrument: Option<&InstrumentAny>,
688 ts_now: UnixNanos,
689) -> Option<OrderEventAny> {
690 let voided_qty = order.leaves_qty();
691 if voided_qty.is_zero() {
692 return None;
693 }
694 let instrument = instrument?;
695 let order_side = order.order_side();
696
697 let last_px = resolve_fill_price(order, report, instrument)
698 .unwrap_or_else(|| Price::zero(instrument.price_precision()));
699
700 let mut event = OrderFillVoided::new(
701 order.trader_id(),
702 order.strategy_id(),
703 order.instrument_id(),
704 order.client_order_id(),
705 report.venue_order_id,
706 report.account_id,
707 Ustr::from(&format!("reconciliation-{}", report.report_id)),
708 TradeId::new(format!("VOID-{}", report.venue_order_id)),
709 voided_qty,
710 None,
711 order_side,
712 order.order_type(),
713 last_px,
714 instrument.quote_currency(),
715 LiquiditySide::NoLiquiditySide,
716 None,
717 report.cancel_reason.as_deref().map(Ustr::from),
718 None,
719 UUID4::new(),
720 report.ts_last,
721 ts_now,
722 true,
723 false,
724 );
725 event.causation_id = Some(report.report_id);
726 Some(OrderEventAny::FillVoided(event))
727}
728
729fn create_external_terminal_event(
730 order: &OrderAny,
731 report: &OrderStatusReport,
732 account_id: AccountId,
733 ts_now: UnixNanos,
734) -> OrderEventAny {
735 match report.order_status {
736 OrderStatus::Canceled => OrderEventAny::Canceled(OrderCanceled::new(
737 order.trader_id(),
738 order.strategy_id(),
739 order.instrument_id(),
740 order.client_order_id(),
741 UUID4::new(),
742 report.ts_last,
743 ts_now,
744 true, Some(report.venue_order_id),
746 Some(account_id),
747 report.cancel_reason.as_deref().map(Ustr::from),
748 )),
749 OrderStatus::Expired => OrderEventAny::Expired(OrderExpired::new(
750 order.trader_id(),
751 order.strategy_id(),
752 order.instrument_id(),
753 order.client_order_id(),
754 UUID4::new(),
755 report.ts_last,
756 ts_now,
757 true, Some(report.venue_order_id),
759 Some(account_id),
760 )),
761 status => unreachable!("cannot create external terminal event for {status}"),
762 }
763}
764
765pub fn reconcile_fill_report(
770 order: &OrderAny,
771 report: &FillReport,
772 instrument: &InstrumentAny,
773 ts_now: UnixNanos,
774 allow_overfills: bool,
775) -> Option<OrderEventAny> {
776 if report.last_qty.is_zero() {
777 log::warn!("Skipping zero-quantity fill report: {report}");
778 return None;
779 }
780
781 if order.trade_ids().iter().any(|id| **id == report.trade_id) {
782 log::debug!(
783 "Duplicate fill detected: trade_id {} already exists for order {}",
784 report.trade_id,
785 order.client_order_id()
786 );
787 return None;
788 }
789
790 let potential_filled_qty = order.filled_qty() + report.last_qty;
791 if potential_filled_qty > order.quantity() {
792 if !allow_overfills {
793 log::warn!(
794 "Rejecting fill that would cause overfill for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, would result in filled_qty={}",
795 order.client_order_id(),
796 order.quantity(),
797 order.filled_qty(),
798 report.last_qty,
799 potential_filled_qty
800 );
801 return None;
802 }
803 log::warn!(
804 "Allowing overfill during reconciliation for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, will result in filled_qty={}",
805 order.client_order_id(),
806 order.quantity(),
807 order.filled_qty(),
808 report.last_qty,
809 potential_filled_qty
810 );
811 }
812
813 let account_id = report.account_id;
814 let venue_order_id = if order.venue_order_id() == Some(report.venue_order_id)
815 || order.venue_order_ids().contains(&&report.venue_order_id)
816 {
817 report.venue_order_id
818 } else {
819 order.venue_order_id().unwrap_or(report.venue_order_id)
820 };
821
822 log::info!(
823 color = LogColor::Blue as u8;
824 "Reconciling fill for {}: qty={}, px={}, trade_id={}",
825 order.client_order_id(),
826 report.last_qty,
827 report.last_px,
828 report.trade_id,
829 );
830
831 Some(OrderEventAny::Filled(OrderFilled::new(
832 order.trader_id(),
833 order.strategy_id(),
834 order.instrument_id(),
835 order.client_order_id(),
836 venue_order_id,
837 account_id,
838 report.trade_id,
839 order.order_side(),
840 order.order_type(),
841 report.last_qty,
842 report.last_px,
843 instrument.quote_currency(),
844 report.liquidity_side,
845 UUID4::new(),
846 report.ts_event,
847 ts_now,
848 true, report.venue_position_id,
850 Some(report.commission),
851 None,
852 )))
853}
854
855pub fn should_reconciliation_update(order: &OrderAny, report: &OrderStatusReport) -> bool {
863 if report.quantity != order.quantity() && report.quantity >= order.filled_qty() {
864 return true;
865 }
866
867 let price_drift = report.price.is_some() && report.price != order.price();
868 let trigger_drift =
869 report.trigger_price.is_some() && report.trigger_price != order.trigger_price();
870
871 match order.order_type() {
872 OrderType::Limit => price_drift,
873 OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched => {
874 trigger_drift
875 }
876 OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched => {
877 trigger_drift || price_drift
878 }
879 _ => false,
880 }
881}
882
883#[must_use]
885pub(super) fn create_reconciliation_accepted(
886 order: &OrderAny,
887 report: &OrderStatusReport,
888 ts_now: UnixNanos,
889) -> Option<OrderEventAny> {
890 let account_id = order.account_id()?;
891
892 Some(OrderEventAny::Accepted(OrderAccepted::new(
893 order.trader_id(),
894 order.strategy_id(),
895 order.instrument_id(),
896 order.client_order_id(),
897 order.venue_order_id().unwrap_or(report.venue_order_id),
898 account_id,
899 UUID4::new(),
900 report.ts_accepted,
901 ts_now,
902 true, )))
904}
905
906#[must_use]
908pub fn create_reconciliation_rejected(
909 order: &OrderAny,
910 reason: Option<&str>,
911 ts_now: UnixNanos,
912) -> Option<OrderEventAny> {
913 let account_id = order.account_id()?;
914 let reason = reason.unwrap_or("UNKNOWN");
915
916 Some(OrderEventAny::Rejected(OrderRejected::new(
917 order.trader_id(),
918 order.strategy_id(),
919 order.instrument_id(),
920 order.client_order_id(),
921 account_id,
922 Ustr::from(reason),
923 UUID4::new(),
924 ts_now,
925 ts_now,
926 true, reason_indicates_post_only_rejection(reason),
928 )))
929}
930
931fn reason_indicates_post_only_rejection(reason: &str) -> bool {
932 let normalized: String = reason
933 .chars()
934 .filter_map(|ch| {
935 if ch == '-' || ch == '_' || ch.is_whitespace() {
936 None
937 } else {
938 Some(ch.to_ascii_lowercase())
939 }
940 })
941 .collect();
942
943 normalized.contains("postonly") || normalized.contains("postwouldexecute")
944}
945
946#[must_use]
948pub fn create_reconciliation_triggered(
949 order: &OrderAny,
950 report: &OrderStatusReport,
951 ts_now: UnixNanos,
952) -> OrderEventAny {
953 OrderEventAny::Triggered(OrderTriggered::new(
954 order.trader_id(),
955 order.strategy_id(),
956 order.instrument_id(),
957 order.client_order_id(),
958 UUID4::new(),
959 report.ts_triggered.unwrap_or(ts_now),
960 ts_now,
961 true, order.venue_order_id(),
963 order.account_id(),
964 ))
965}
966
967#[must_use]
969pub(super) fn create_reconciliation_canceled(
970 order: &OrderAny,
971 report: &OrderStatusReport,
972 ts_now: UnixNanos,
973) -> OrderEventAny {
974 OrderEventAny::Canceled(OrderCanceled::new(
975 order.trader_id(),
976 order.strategy_id(),
977 order.instrument_id(),
978 order.client_order_id(),
979 UUID4::new(),
980 report.ts_last,
981 ts_now,
982 true, order.venue_order_id(),
984 order.account_id(),
985 report.cancel_reason.as_deref().map(Ustr::from),
986 ))
987}
988
989#[must_use]
991pub(super) fn create_reconciliation_expired(
992 order: &OrderAny,
993 report: &OrderStatusReport,
994 ts_now: UnixNanos,
995) -> OrderEventAny {
996 OrderEventAny::Expired(OrderExpired::new(
997 order.trader_id(),
998 order.strategy_id(),
999 order.instrument_id(),
1000 order.client_order_id(),
1001 UUID4::new(),
1002 report.ts_last,
1003 ts_now,
1004 true, order.venue_order_id(),
1006 order.account_id(),
1007 ))
1008}
1009
1010#[must_use]
1012pub(super) fn create_reconciliation_updated(
1013 order: &OrderAny,
1014 report: &OrderStatusReport,
1015 ts_now: UnixNanos,
1016) -> OrderEventAny {
1017 let trigger_price = match order.order_type() {
1024 OrderType::StopMarket
1025 | OrderType::StopLimit
1026 | OrderType::MarketIfTouched
1027 | OrderType::LimitIfTouched
1028 | OrderType::TrailingStopMarket
1029 | OrderType::TrailingStopLimit => report.trigger_price,
1030 _ => None,
1031 };
1032
1033 OrderEventAny::Updated(OrderUpdated::new(
1034 order.trader_id(),
1035 order.strategy_id(),
1036 order.instrument_id(),
1037 order.client_order_id(),
1038 report.quantity,
1039 UUID4::new(),
1040 report.ts_last,
1041 ts_now,
1042 true, order.venue_order_id(),
1044 order.account_id(),
1045 report.price,
1046 trigger_price,
1047 None, order.is_quote_quantity(),
1049 ))
1050}
1051
1052pub(super) fn create_inferred_fill(
1054 order: &OrderAny,
1055 report: &OrderStatusReport,
1056 account_id: AccountId,
1057 instrument: &InstrumentAny,
1058 ts_now: UnixNanos,
1059 commission: Option<Money>,
1060) -> Option<OrderEventAny> {
1061 let cached_order_side = order.order_side();
1062 let order_side = report.order_side.unwrap_or(cached_order_side);
1063 if order_side != cached_order_side {
1064 log::warn!(
1065 "Order side mismatch for {}: cached={:?}, venue={order_side:?}",
1066 order.client_order_id(),
1067 cached_order_side,
1068 );
1069 }
1070
1071 let liquidity_side = match order.order_type() {
1072 OrderType::Market
1073 | OrderType::StopMarket
1074 | OrderType::MarketToLimit
1075 | OrderType::TrailingStopMarket => LiquiditySide::Taker,
1076 _ if order.is_post_only() => LiquiditySide::Maker,
1077 _ => LiquiditySide::NoLiquiditySide,
1078 };
1079
1080 let Some(last_px) = resolve_fill_price(order, report, instrument) else {
1081 log::warn!(
1082 "Cannot create inferred fill for {}: no avg_px, report price, or order price",
1083 order.client_order_id()
1084 );
1085
1086 return None;
1087 };
1088 let last_px = clamp_inferred_fill_price(last_px, instrument);
1089
1090 let position_id = reconciliation_position_id(report, instrument);
1091 let trade_id = create_inferred_reconciliation_trade_id(
1092 account_id,
1093 order.instrument_id(),
1094 order.client_order_id(),
1095 Some(report.venue_order_id),
1096 order_side,
1097 order.order_type(),
1098 report.filled_qty,
1099 report.filled_qty,
1100 last_px,
1101 position_id,
1102 report.ts_last,
1103 );
1104
1105 log::info!(
1106 "Generated inferred fill for {} ({}) qty={} px={}",
1107 order.client_order_id(),
1108 report.venue_order_id,
1109 report.filled_qty,
1110 last_px,
1111 );
1112
1113 Some(OrderEventAny::Filled(OrderFilled::new(
1114 order.trader_id(),
1115 order.strategy_id(),
1116 order.instrument_id(),
1117 order.client_order_id(),
1118 report.venue_order_id,
1119 account_id,
1120 trade_id,
1121 order_side,
1122 order.order_type(),
1123 report.filled_qty,
1124 last_px,
1125 instrument.quote_currency(),
1126 liquidity_side,
1127 UUID4::new(),
1128 report.ts_last,
1129 ts_now,
1130 true, report.venue_position_id,
1132 commission,
1133 None,
1134 )))
1135}
1136
1137pub fn create_incremental_inferred_fill(
1139 order: &OrderAny,
1140 report: &OrderStatusReport,
1141 account_id: &AccountId,
1142 instrument: &InstrumentAny,
1143 ts_now: UnixNanos,
1144 commission: Option<Money>,
1145) -> Option<OrderEventAny> {
1146 let order_side = order.order_side();
1147
1148 let order_filled_qty = order.filled_qty();
1149 debug_assert!(
1150 report.filled_qty >= order_filled_qty,
1151 "incremental inferred fill requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
1152 report.filled_qty,
1153 order_filled_qty,
1154 order.client_order_id(),
1155 );
1156 let last_qty = report.filled_qty - order_filled_qty;
1157
1158 if last_qty <= Quantity::zero(instrument.size_precision()) {
1159 return None;
1160 }
1161
1162 let (last_px, liquidity_side) =
1163 incremental_inferred_fill_price_and_liquidity(order, report, instrument)?;
1164
1165 let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
1166 let position_id = reconciliation_position_id(report, instrument);
1167 let trade_id = create_inferred_reconciliation_trade_id(
1168 *account_id,
1169 order.instrument_id(),
1170 order.client_order_id(),
1171 Some(venue_order_id),
1172 order_side,
1173 order.order_type(),
1174 report.filled_qty,
1175 last_qty,
1176 last_px,
1177 position_id,
1178 report.ts_last,
1179 );
1180
1181 log::info!(
1182 color = LogColor::Blue as u8;
1183 "Generated inferred fill for {}: qty={}, px={}",
1184 order.client_order_id(),
1185 last_qty,
1186 last_px,
1187 );
1188
1189 Some(OrderEventAny::Filled(OrderFilled::new(
1190 order.trader_id(),
1191 order.strategy_id(),
1192 order.instrument_id(),
1193 order.client_order_id(),
1194 venue_order_id,
1195 *account_id,
1196 trade_id,
1197 order_side,
1198 order.order_type(),
1199 last_qty,
1200 last_px,
1201 instrument.quote_currency(),
1202 liquidity_side,
1203 UUID4::new(),
1204 report.ts_last,
1205 ts_now,
1206 true, report.venue_position_id,
1208 commission,
1209 None,
1210 )))
1211}
1212
1213#[must_use]
1221pub fn incremental_inferred_fill_price_and_liquidity(
1222 order: &OrderAny,
1223 report: &OrderStatusReport,
1224 instrument: &InstrumentAny,
1225) -> Option<(Price, LiquiditySide)> {
1226 let last_px = calculate_incremental_fill_price(order, report, instrument)?;
1227
1228 Some((
1229 clamp_inferred_fill_price(last_px, instrument),
1230 inferred_fill_liquidity_side(order),
1231 ))
1232}
1233
1234#[must_use]
1241pub fn inferred_fill_price_and_liquidity(
1242 order: &OrderAny,
1243 report: &OrderStatusReport,
1244 instrument: &InstrumentAny,
1245) -> Option<(Price, LiquiditySide)> {
1246 let last_px = resolve_fill_price(order, report, instrument)?;
1247
1248 Some((
1249 clamp_inferred_fill_price(last_px, instrument),
1250 inferred_fill_liquidity_side(order),
1251 ))
1252}
1253
1254fn inferred_fill_liquidity_side(order: &OrderAny) -> LiquiditySide {
1255 match order.order_type() {
1256 OrderType::Market
1257 | OrderType::StopMarket
1258 | OrderType::MarketToLimit
1259 | OrderType::TrailingStopMarket => LiquiditySide::Taker,
1260 _ if order.is_post_only() => LiquiditySide::Maker,
1261 _ => LiquiditySide::NoLiquiditySide,
1262 }
1263}
1264
1265pub fn create_inferred_fill_for_qty(
1271 order: &OrderAny,
1272 report: &OrderStatusReport,
1273 account_id: &AccountId,
1274 instrument: &InstrumentAny,
1275 fill_qty: Quantity,
1276 ts_now: UnixNanos,
1277 commission: Option<Money>,
1278) -> Option<OrderEventAny> {
1279 if fill_qty.is_zero() {
1280 return None;
1281 }
1282
1283 let order_side = order.order_side();
1284
1285 let Some((last_px, liquidity_side)) =
1286 inferred_fill_price_and_liquidity(order, report, instrument)
1287 else {
1288 log::warn!(
1289 "Cannot determine fill price for {}: no avg_px, report price, or order price",
1290 order.client_order_id()
1291 );
1292
1293 return None;
1294 };
1295
1296 let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
1297 let position_id = reconciliation_position_id(report, instrument);
1298 let trade_id = create_inferred_reconciliation_trade_id(
1299 *account_id,
1300 order.instrument_id(),
1301 order.client_order_id(),
1302 Some(venue_order_id),
1303 order_side,
1304 order.order_type(),
1305 report.filled_qty,
1306 fill_qty,
1307 last_px,
1308 position_id,
1309 report.ts_last,
1310 );
1311
1312 log::info!(
1313 color = LogColor::Blue as u8;
1314 "Generated inferred fill for {}: qty={}, px={}",
1315 order.client_order_id(),
1316 fill_qty,
1317 last_px,
1318 );
1319
1320 Some(OrderEventAny::Filled(OrderFilled::new(
1321 order.trader_id(),
1322 order.strategy_id(),
1323 order.instrument_id(),
1324 order.client_order_id(),
1325 venue_order_id,
1326 *account_id,
1327 trade_id,
1328 order_side,
1329 order.order_type(),
1330 fill_qty,
1331 last_px,
1332 instrument.quote_currency(),
1333 liquidity_side,
1334 UUID4::new(),
1335 report.ts_last,
1336 ts_now,
1337 true, report.venue_position_id,
1339 commission,
1340 None,
1341 )))
1342}
1343
1344fn report_is_confirmed_state(report: &OrderStatusReport) -> bool {
1345 matches!(
1346 report.order_status,
1347 OrderStatus::Accepted
1348 | OrderStatus::Triggered
1349 | OrderStatus::PartiallyFilled
1350 | OrderStatus::Filled
1351 | OrderStatus::Canceled
1352 | OrderStatus::Expired
1353 )
1354}
1355
1356fn report_is_working(report: &OrderStatusReport) -> bool {
1357 matches!(
1358 report.order_status,
1359 OrderStatus::Accepted | OrderStatus::Triggered | OrderStatus::PartiallyFilled
1360 )
1361}
1362
1363pub(crate) fn is_superseded_terminal_report(order: &OrderAny, report: &OrderStatusReport) -> bool {
1364 if !matches!(
1365 report.order_status,
1366 OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Filled
1367 ) {
1368 return false;
1369 }
1370
1371 let Some(cached_venue_order_id) = order.venue_order_id() else {
1372 return false;
1373 };
1374
1375 cached_venue_order_id != report.venue_order_id
1376 && order
1377 .venue_order_ids()
1378 .iter()
1379 .any(|venue_order_id| **venue_order_id == report.venue_order_id)
1380}
1381
1382fn local_accepts_amendment(order: &OrderAny) -> bool {
1383 matches!(
1384 order.status(),
1385 OrderStatus::Accepted | OrderStatus::Triggered | OrderStatus::PartiallyFilled
1386 )
1387}
1388
1389fn pending_terminal_quantity_update(order: &OrderAny, report: &OrderStatusReport) -> bool {
1390 if !matches!(
1391 order.status(),
1392 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
1393 ) {
1394 return false;
1395 }
1396
1397 if matches!(
1398 report.order_status,
1399 OrderStatus::Canceled | OrderStatus::Expired
1400 ) {
1401 return true;
1402 }
1403
1404 report.order_status == OrderStatus::Filled
1405 && report.filled_qty > order.quantity()
1406 && report.quantity >= report.filled_qty
1407}
1408
1409fn should_accept_before_reconciliation(order: &OrderAny, report: &OrderStatusReport) -> bool {
1410 order.status() == OrderStatus::Submitted && report.order_status != OrderStatus::Rejected
1411}
1412
1413fn reconcile_fill_quantity_mismatch(
1417 order: &OrderAny,
1418 report: &OrderStatusReport,
1419 instrument: Option<&InstrumentAny>,
1420 ts_now: UnixNanos,
1421 commission: Option<Money>,
1422) -> Option<OrderEventAny> {
1423 let order_filled_qty = order.filled_qty();
1424 let report_filled_qty = report.filled_qty;
1425
1426 if report_filled_qty < order_filled_qty {
1427 let precision = order_filled_qty.precision.max(report_filled_qty.precision);
1430 if is_within_single_unit_tolerance(
1431 report_filled_qty.as_decimal(),
1432 order_filled_qty.as_decimal(),
1433 precision,
1434 ) {
1435 return None;
1436 }
1437
1438 log::warn!(
1439 "Fill qty mismatch for {} ({}): cached={}, venue={}, order_qty={} (venue < cached)",
1440 order.client_order_id(),
1441 report.venue_order_id,
1442 order_filled_qty,
1443 report_filled_qty,
1444 order.quantity(),
1445 );
1446 return None;
1447 }
1448
1449 if report_filled_qty > order_filled_qty {
1450 if order.is_closed() {
1453 let precision = order_filled_qty.precision.max(report_filled_qty.precision);
1454
1455 if is_within_single_unit_tolerance(
1456 report_filled_qty.as_decimal(),
1457 order_filled_qty.as_decimal(),
1458 precision,
1459 ) {
1460 return None;
1461 }
1462
1463 log::debug!(
1464 "{} {} already closed but reported difference in filled_qty: \
1465 report={}, cached={}, skipping inferred fill generation for closed order",
1466 order.instrument_id(),
1467 order.client_order_id(),
1468 report_filled_qty,
1469 order_filled_qty,
1470 );
1471 return None;
1472 }
1473
1474 let Some(instrument) = instrument else {
1476 log::warn!(
1477 "Cannot generate inferred fill for {}: instrument not available",
1478 order.client_order_id()
1479 );
1480 return None;
1481 };
1482
1483 let account_id = order.account_id()?;
1484 return create_incremental_inferred_fill(
1485 order,
1486 report,
1487 &account_id,
1488 instrument,
1489 ts_now,
1490 commission,
1491 );
1492 }
1493
1494 if order.status() != report.order_status {
1499 if should_reconciliation_update(order, report) {
1500 log::info!(
1501 "Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, \
1502 filled_qty={}, updating quantity {}->{}",
1503 order.client_order_id(),
1504 order.status(),
1505 report.order_status,
1506 report.filled_qty,
1507 order.quantity(),
1508 report.quantity,
1509 );
1510 return Some(create_reconciliation_updated(order, report, ts_now));
1511 }
1512
1513 log::warn!(
1514 "Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, filled_qty={}",
1515 order.client_order_id(),
1516 order.status(),
1517 report.order_status,
1518 report.filled_qty
1519 );
1520 }
1521
1522 None
1523}
1524
1525fn calculate_incremental_fill_price(
1531 order: &OrderAny,
1532 report: &OrderStatusReport,
1533 instrument: &InstrumentAny,
1534) -> Option<Price> {
1535 let order_filled_qty = order.filled_qty();
1536 debug_assert!(
1537 report.filled_qty >= order_filled_qty,
1538 "incremental fill price requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
1539 report.filled_qty,
1540 order_filled_qty,
1541 order.client_order_id(),
1542 );
1543
1544 if order_filled_qty.is_zero() {
1546 let last_px = resolve_fill_price(order, report, instrument);
1547 if last_px.is_none() {
1548 log::warn!(
1549 "Cannot determine fill price for {}: no avg_px, report price, or order price",
1550 order.client_order_id()
1551 );
1552 }
1553
1554 return last_px;
1555 }
1556
1557 if let Some(report_avg_px) = report.avg_px {
1559 let last_px_decimal = match order.avg_px() {
1560 None => report_avg_px,
1562 Some(order_avg_px) => {
1563 let report_filled_qty = report.filled_qty;
1564 let last_qty = report_filled_qty - order_filled_qty;
1565
1566 let report_notional = report_avg_px * report_filled_qty.as_decimal();
1567 let order_notional = order_avg_px * order_filled_qty.as_decimal();
1568 let last_notional = report_notional - order_notional;
1569 let back_solved = last_notional / last_qty.as_decimal();
1570
1571 if back_solved < Decimal::ZERO && !instrument.allows_negative_price() {
1572 if report_avg_px < Decimal::ZERO {
1573 log::warn!(
1574 "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",
1575 order.client_order_id(),
1576 );
1577
1578 return None;
1579 }
1580
1581 log::warn!(
1582 "Negative back-solved fill price {back_solved} for {}, using venue average {report_avg_px}",
1583 order.client_order_id(),
1584 );
1585
1586 report_avg_px
1587 } else {
1588 back_solved
1589 }
1590 }
1591 };
1592
1593 return reconciliation_price(last_px_decimal, instrument)
1594 .inspect_err(|e| {
1595 log::warn!(
1596 "Cannot price {} from incremental {last_px_decimal}, falling back: {e}",
1597 order.client_order_id(),
1598 );
1599 })
1600 .ok()
1601 .or_else(|| resolve_fill_price(order, report, instrument));
1602 }
1603
1604 resolve_fill_price(order, report, instrument)
1605}
1606
1607fn resolve_fill_price(
1613 order: &OrderAny,
1614 report: &OrderStatusReport,
1615 instrument: &InstrumentAny,
1616) -> Option<Price> {
1617 report
1618 .avg_px
1619 .and_then(|avg_px| {
1620 reconciliation_price(avg_px, instrument)
1621 .inspect_err(|e| {
1622 log::warn!(
1623 "Cannot price {} from venue average {avg_px}, trying next source: {e}",
1624 order.client_order_id(),
1625 );
1626 })
1627 .ok()
1628 })
1629 .or(report.price)
1630 .or_else(|| order.price())
1631}
1632
1633fn clamp_inferred_fill_price(price: Price, instrument: &InstrumentAny) -> Price {
1638 let px = cap_price_at_instrument_max(price.as_decimal(), instrument);
1639 Price::from_decimal_dp(px, price.precision.max(instrument.price_precision())).unwrap_or(price)
1640}
1641
1642fn reconciliation_position_id(
1643 report: &OrderStatusReport,
1644 instrument: &InstrumentAny,
1645) -> PositionId {
1646 report
1647 .venue_position_id
1648 .unwrap_or_else(|| PositionId::new(format!("{}-EXTERNAL", instrument.id())))
1649}