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::{LIMIT_ORDER_TYPES, 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 is_working(report.order_status) && 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 if report.order_status == OrderStatus::Triggered
257 && working.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 && (is_working(working.status())
295 || working.status() == OrderStatus::PendingUpdate && is_working(report.order_status)
296 || pending_terminal_quantity_update(&working, report))
297 && should_reconciliation_update(&working, report)
298 {
299 let updated = create_reconciliation_updated(&working, report, ts_now);
300 if let Err(e) = working.apply(updated.clone()) {
301 log::warn!(
302 "Failed to pre-apply reconciliation update for {}: {e}",
303 order.client_order_id(),
304 );
305 } else {
306 events.push(updated);
307 }
308 }
309
310 (working, events)
311}
312
313#[must_use]
327pub fn reconcile_order_report(
328 order: &OrderAny,
329 report: &OrderStatusReport,
330 instrument: Option<&InstrumentAny>,
331 ts_now: UnixNanos,
332) -> Option<OrderEventAny> {
333 reconcile_order_report_with_commission(order, report, instrument, ts_now, None)
334}
335
336#[must_use]
338pub fn reconcile_order_report_with_commission(
339 order: &OrderAny,
340 report: &OrderStatusReport,
341 instrument: Option<&InstrumentAny>,
342 ts_now: UnixNanos,
343 commission: Option<Money>,
344) -> Option<OrderEventAny> {
345 if matches!(
346 report.order_status,
347 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
348 ) {
349 log::debug!(
350 "Order {} venue report in pending state: {:?}",
351 order.client_order_id(),
352 report.order_status,
353 );
354 return None;
355 }
356
357 if is_unchanged_accepted_report_during_pending_command(order, report) {
358 log::debug!(
359 "Order {} remains inflight while venue reports an unchanged accepted snapshot",
360 order.client_order_id(),
361 );
362 return None;
363 }
364
365 if is_superseded_terminal_report(order, report) {
366 let cached_venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
367 log::info!(
368 "Suppressing {:?} for {} on previously-promoted venue_order_id {}: \
369 current venue_order_id is {}",
370 report.order_status,
371 order.client_order_id(),
372 report.venue_order_id,
373 cached_venue_order_id,
374 );
375 return None;
376 }
377
378 if order.status() == report.order_status && order.filled_qty() == report.filled_qty {
379 if should_reconciliation_update(order, report) {
380 log::info!(
381 "Order {} has been updated at venue: qty={}->{}, price={:?}->{:?}",
382 order.client_order_id(),
383 order.quantity(),
384 report.quantity,
385 order.price(),
386 report.price
387 );
388 return Some(create_reconciliation_updated(order, report, ts_now));
389 }
390 return None; }
392
393 match report.order_status {
394 OrderStatus::Accepted => {
395 if matches!(
396 order.status(),
397 OrderStatus::Accepted | OrderStatus::PendingUpdate
398 ) && should_reconciliation_update(order, report)
399 {
400 return Some(create_reconciliation_updated(order, report, ts_now));
401 }
402 create_reconciliation_accepted(order, report, ts_now)
403 }
404 OrderStatus::Rejected => {
405 create_reconciliation_rejected(order, report.cancel_reason.as_deref(), ts_now)
406 }
407 OrderStatus::Triggered => {
408 if TRIGGERABLE_ORDER_TYPES.contains(&order.order_type()) {
409 Some(create_reconciliation_triggered(order, report, ts_now))
410 } else {
411 log::debug!(
412 "Skipping OrderTriggered for {} order {}: market-style stops have no TRIGGERED state",
413 order.order_type(),
414 order.client_order_id(),
415 );
416 None
417 }
418 }
419 OrderStatus::Canceled => Some(create_reconciliation_canceled(order, report, ts_now)),
420 OrderStatus::Expired => Some(create_reconciliation_expired(order, report, ts_now)),
421
422 OrderStatus::PartiallyFilled | OrderStatus::Filled => {
423 reconcile_fill_quantity_mismatch(order, report, instrument, ts_now, commission)
424 }
425
426 OrderStatus::Voided => {
427 create_reconciliation_terminal_fill_void(order, report, instrument, ts_now)
428 }
429
430 OrderStatus::PendingUpdate | OrderStatus::PendingCancel => None,
431
432 OrderStatus::Initialized
434 | OrderStatus::Submitted
435 | OrderStatus::Denied
436 | OrderStatus::Emulated
437 | OrderStatus::Released => {
438 log::warn!(
439 "Unexpected order status in venue report for {}: {:?}",
440 order.client_order_id(),
441 report.order_status
442 );
443 None
444 }
445 }
446}
447
448fn is_unchanged_accepted_report_during_pending_command(
449 order: &OrderAny,
450 report: &OrderStatusReport,
451) -> bool {
452 matches!(
453 order.status(),
454 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
455 ) && report.order_status == OrderStatus::Accepted
456 && order.venue_order_id() == Some(report.venue_order_id)
457 && order.filled_qty() == report.filled_qty
458 && !should_reconciliation_update(order, report)
459}
460
461pub fn generate_external_order_status_events(
468 order: &OrderAny,
469 report: &OrderStatusReport,
470 account_id: &AccountId,
471 instrument: &InstrumentAny,
472 ts_now: UnixNanos,
473) -> Vec<OrderEventAny> {
474 generate_external_order_status_events_with_commission(
475 order, report, account_id, instrument, ts_now, None,
476 )
477}
478
479#[must_use]
481pub fn generate_external_order_status_events_with_commission(
482 order: &OrderAny,
483 report: &OrderStatusReport,
484 account_id: &AccountId,
485 instrument: &InstrumentAny,
486 ts_now: UnixNanos,
487 commission: Option<Money>,
488) -> Vec<OrderEventAny> {
489 let accepted = OrderEventAny::Accepted(OrderAccepted::new(
490 order.trader_id(),
491 order.strategy_id(),
492 order.instrument_id(),
493 order.client_order_id(),
494 report.venue_order_id,
495 *account_id,
496 UUID4::new(),
497 report.ts_accepted,
498 ts_now,
499 true, ));
501
502 match report.order_status {
503 OrderStatus::Accepted | OrderStatus::Triggered => vec![accepted],
504 OrderStatus::PartiallyFilled | OrderStatus::Filled => {
505 let mut events = vec![accepted];
506
507 if !report.filled_qty.is_zero()
508 && let Some(filled) =
509 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
510 {
511 events.push(filled);
512 }
513
514 events
515 }
516 OrderStatus::Voided => {
517 let mut working = order.clone();
518 if let Err(e) = working.apply(accepted.clone()) {
519 log::warn!(
520 "Cannot project external order acceptance for {}: {e}",
521 order.client_order_id()
522 );
523 return vec![accepted];
524 }
525 let mut events = vec![accepted];
526
527 if !report.filled_qty.is_zero()
528 && let Some(filled) =
529 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
530 {
531 if let Err(e) = working.apply(filled.clone()) {
532 log::warn!(
533 "Cannot project external order fill for {}: {e}",
534 order.client_order_id()
535 );
536 } else {
537 events.push(filled);
538 }
539 }
540
541 if let Some(voided) =
542 create_reconciliation_terminal_fill_void(&working, report, Some(instrument), ts_now)
543 {
544 events.push(voided);
545 }
546 events
547 }
548 OrderStatus::Canceled | OrderStatus::Expired => {
549 let terminal = create_external_terminal_event(order, report, *account_id, ts_now);
550 let mut events = vec![accepted];
551
552 let inferred_fill = if report.filled_qty.is_zero() {
553 None
554 } else {
555 create_inferred_fill(order, report, *account_id, instrument, ts_now, commission)
556 };
557 let filled_to_quantity =
558 inferred_fill.is_some() && report.filled_qty >= report.quantity;
559 if let Some(filled) = inferred_fill {
560 events.push(filled);
561 }
562
563 if !filled_to_quantity {
564 events.push(terminal);
565 }
566 events
567 }
568 OrderStatus::Rejected => {
569 let reason = report.cancel_reason.as_deref().unwrap_or("UNKNOWN");
571 vec![OrderEventAny::Rejected(OrderRejected::new(
572 order.trader_id(),
573 order.strategy_id(),
574 order.instrument_id(),
575 order.client_order_id(),
576 *account_id,
577 Ustr::from(reason),
578 UUID4::new(),
579 report.ts_last,
580 ts_now,
581 true, reason_indicates_post_only_rejection(reason),
583 ))]
584 }
585 _ => {
586 log::warn!(
587 "Unhandled order status {} for external order {}",
588 report.order_status,
589 order.client_order_id()
590 );
591 Vec::new()
592 }
593 }
594}
595
596fn create_reconciliation_fill_voids(
597 order: &OrderAny,
598 report: &OrderStatusReport,
599 ts_now: UnixNanos,
600) -> Vec<OrderEventAny> {
601 let mut remaining = order.filled_qty() - report.filled_qty;
602 let order_events = order.events();
603 let mut corrections = Vec::new();
604
605 let mut seen = HashSet::new();
608
609 let quantity_fills: Vec<&OrderFilled> = order_events
610 .iter()
611 .filter_map(|event| match event {
612 OrderEventAny::Filled(fill) if seen.insert(fill.trade_id) => Some(fill),
613 _ => None,
614 })
615 .collect();
616
617 for fill in quantity_fills.iter().rev() {
618 if remaining.is_zero() {
619 break;
620 }
621 let previous = order_events.iter().rev().find_map(|event| match event {
622 OrderEventAny::FillVoided(voided) if voided.trade_id == fill.trade_id => Some(voided),
623 _ => None,
624 });
625 let prior_qty = previous.map_or_else(
626 || Quantity::zero(fill.last_qty.precision),
627 |voided| voided.voided_qty.min(fill.last_qty),
628 );
629 let effective = fill.last_qty - prior_qty;
630 if effective.is_zero() {
631 continue;
632 }
633 let removed = remaining.min(effective);
634 let voided_qty = prior_qty + removed;
635 let commission_voided = fill.commission.and_then(|commission| {
636 let fraction = voided_qty.as_decimal() / fill.last_qty.as_decimal();
637 Money::from_decimal(commission.as_decimal() * fraction, commission.currency).ok()
638 });
639 let mut event = OrderFillVoided::new(
640 fill.trader_id,
641 fill.strategy_id,
642 fill.instrument_id,
643 fill.client_order_id,
644 fill.venue_order_id,
645 fill.account_id,
646 Ustr::from(&format!(
647 "reconciliation-{}-{}",
648 report.report_id, fill.trade_id
649 )),
650 fill.trade_id,
651 voided_qty,
652 commission_voided,
653 fill.order_side,
654 fill.order_type,
655 fill.last_px,
656 fill.currency,
657 fill.liquidity_side,
658 fill.position_id,
659 report.cancel_reason.as_deref().map(Ustr::from),
660 None,
661 UUID4::new(),
662 report.ts_last,
663 ts_now,
664 true,
665 is_working(report.order_status),
666 );
667 event.causation_id = Some(report.report_id);
668 corrections.push(OrderEventAny::FillVoided(event));
669 remaining = remaining - removed;
670 }
671
672 if !remaining.is_zero() {
673 log::warn!(
674 "Cannot reconcile fill decrease for {}: {} is outside retained fill history",
675 order.client_order_id(),
676 remaining
677 );
678 return Vec::new();
679 }
680 corrections
681}
682
683fn create_reconciliation_terminal_fill_void(
689 order: &OrderAny,
690 report: &OrderStatusReport,
691 instrument: Option<&InstrumentAny>,
692 ts_now: UnixNanos,
693) -> Option<OrderEventAny> {
694 let voided_qty = order.leaves_qty();
695 if voided_qty.is_zero() {
696 return None;
697 }
698 let instrument = instrument?;
699 let order_side = order.order_side();
700
701 let last_px = resolve_fill_price(order, report, instrument)
702 .unwrap_or_else(|| Price::zero(instrument.price_precision()));
703
704 let mut event = OrderFillVoided::new(
705 order.trader_id(),
706 order.strategy_id(),
707 order.instrument_id(),
708 order.client_order_id(),
709 report.venue_order_id,
710 report.account_id,
711 Ustr::from(&format!("reconciliation-{}", report.report_id)),
712 TradeId::new(format!("VOID-{}", report.venue_order_id)),
713 voided_qty,
714 None,
715 order_side,
716 order.order_type(),
717 last_px,
718 instrument.quote_currency(),
719 LiquiditySide::NoLiquiditySide,
720 None,
721 report.cancel_reason.as_deref().map(Ustr::from),
722 None,
723 UUID4::new(),
724 report.ts_last,
725 ts_now,
726 true,
727 false,
728 );
729 event.causation_id = Some(report.report_id);
730 Some(OrderEventAny::FillVoided(event))
731}
732
733fn create_external_terminal_event(
734 order: &OrderAny,
735 report: &OrderStatusReport,
736 account_id: AccountId,
737 ts_now: UnixNanos,
738) -> OrderEventAny {
739 match report.order_status {
740 OrderStatus::Canceled => OrderEventAny::Canceled(OrderCanceled::new(
741 order.trader_id(),
742 order.strategy_id(),
743 order.instrument_id(),
744 order.client_order_id(),
745 UUID4::new(),
746 report.ts_last,
747 ts_now,
748 true, Some(report.venue_order_id),
750 Some(account_id),
751 report.cancel_reason.as_deref().map(Ustr::from),
752 )),
753 OrderStatus::Expired => OrderEventAny::Expired(OrderExpired::new(
754 order.trader_id(),
755 order.strategy_id(),
756 order.instrument_id(),
757 order.client_order_id(),
758 UUID4::new(),
759 report.ts_last,
760 ts_now,
761 true, Some(report.venue_order_id),
763 Some(account_id),
764 )),
765 status => unreachable!("cannot create external terminal event for {status}"),
766 }
767}
768
769pub fn reconcile_fill_report(
774 order: &OrderAny,
775 report: &FillReport,
776 instrument: &InstrumentAny,
777 ts_now: UnixNanos,
778 allow_overfills: bool,
779) -> Option<OrderEventAny> {
780 if report.last_qty.is_zero() {
781 log::warn!("Skipping zero-quantity fill report: {report}");
782 return None;
783 }
784
785 if order.trade_ids().iter().any(|id| **id == report.trade_id) {
786 log::debug!(
787 "Duplicate fill detected: trade_id {} already exists for order {}",
788 report.trade_id,
789 order.client_order_id()
790 );
791 return None;
792 }
793
794 let potential_filled_qty = order.filled_qty() + report.last_qty;
795 if potential_filled_qty > order.quantity() {
796 if !allow_overfills {
797 log::warn!(
798 "Rejecting fill that would cause overfill for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, would result in filled_qty={}",
799 order.client_order_id(),
800 order.quantity(),
801 order.filled_qty(),
802 report.last_qty,
803 potential_filled_qty
804 );
805 return None;
806 }
807 log::warn!(
808 "Allowing overfill during reconciliation for {}: order.quantity={}, order.filled_qty={}, fill.last_qty={}, will result in filled_qty={}",
809 order.client_order_id(),
810 order.quantity(),
811 order.filled_qty(),
812 report.last_qty,
813 potential_filled_qty
814 );
815 }
816
817 let account_id = report.account_id;
818 let venue_order_id = if order.venue_order_id() == Some(report.venue_order_id)
819 || order.venue_order_ids().contains(&&report.venue_order_id)
820 {
821 report.venue_order_id
822 } else {
823 order.venue_order_id().unwrap_or(report.venue_order_id)
824 };
825
826 log::info!(
827 color = LogColor::Blue as u8;
828 "Reconciling fill for {}: qty={}, px={}, trade_id={}",
829 order.client_order_id(),
830 report.last_qty,
831 report.last_px,
832 report.trade_id,
833 );
834
835 Some(OrderEventAny::Filled(OrderFilled::new(
836 order.trader_id(),
837 order.strategy_id(),
838 order.instrument_id(),
839 order.client_order_id(),
840 venue_order_id,
841 account_id,
842 report.trade_id,
843 order.order_side(),
844 order.order_type(),
845 report.last_qty,
846 report.last_px,
847 instrument.quote_currency(),
848 report.liquidity_side,
849 UUID4::new(),
850 report.ts_event,
851 ts_now,
852 true, report.venue_position_id,
854 Some(report.commission),
855 None,
856 )))
857}
858
859pub fn should_reconciliation_update(order: &OrderAny, report: &OrderStatusReport) -> bool {
867 if report.quantity != order.quantity() && report.quantity >= order.filled_qty() {
868 return true;
869 }
870
871 let price_drift = report.price.is_some() && report.price != order.price();
872 let trigger_drift =
873 report.trigger_price.is_some() && report.trigger_price != order.trigger_price();
874
875 match order.order_type() {
876 OrderType::Limit => price_drift,
877 OrderType::StopMarket | OrderType::TrailingStopMarket | OrderType::MarketIfTouched => {
878 trigger_drift
879 }
880 OrderType::StopLimit | OrderType::TrailingStopLimit | OrderType::LimitIfTouched => {
881 trigger_drift || price_drift
882 }
883 _ => false,
884 }
885}
886
887#[must_use]
889pub(super) fn create_reconciliation_accepted(
890 order: &OrderAny,
891 report: &OrderStatusReport,
892 ts_now: UnixNanos,
893) -> Option<OrderEventAny> {
894 let account_id = order.account_id()?;
895
896 Some(OrderEventAny::Accepted(OrderAccepted::new(
897 order.trader_id(),
898 order.strategy_id(),
899 order.instrument_id(),
900 order.client_order_id(),
901 order.venue_order_id().unwrap_or(report.venue_order_id),
902 account_id,
903 UUID4::new(),
904 report.ts_accepted,
905 ts_now,
906 true, )))
908}
909
910#[must_use]
912pub fn create_reconciliation_rejected(
913 order: &OrderAny,
914 reason: Option<&str>,
915 ts_now: UnixNanos,
916) -> Option<OrderEventAny> {
917 let account_id = order.account_id()?;
918 let reason = reason.unwrap_or("UNKNOWN");
919
920 Some(OrderEventAny::Rejected(OrderRejected::new(
921 order.trader_id(),
922 order.strategy_id(),
923 order.instrument_id(),
924 order.client_order_id(),
925 account_id,
926 Ustr::from(reason),
927 UUID4::new(),
928 ts_now,
929 ts_now,
930 true, reason_indicates_post_only_rejection(reason),
932 )))
933}
934
935fn reason_indicates_post_only_rejection(reason: &str) -> bool {
936 let normalized: String = reason
937 .chars()
938 .filter_map(|ch| {
939 if ch == '-' || ch == '_' || ch.is_whitespace() {
940 None
941 } else {
942 Some(ch.to_ascii_lowercase())
943 }
944 })
945 .collect();
946
947 normalized.contains("postonly") || normalized.contains("postwouldexecute")
948}
949
950#[must_use]
952pub fn create_reconciliation_triggered(
953 order: &OrderAny,
954 report: &OrderStatusReport,
955 ts_now: UnixNanos,
956) -> OrderEventAny {
957 OrderEventAny::Triggered(OrderTriggered::new(
958 order.trader_id(),
959 order.strategy_id(),
960 order.instrument_id(),
961 order.client_order_id(),
962 UUID4::new(),
963 report.ts_triggered.unwrap_or(ts_now),
964 ts_now,
965 true, order.venue_order_id(),
967 order.account_id(),
968 ))
969}
970
971#[must_use]
973pub(super) fn create_reconciliation_canceled(
974 order: &OrderAny,
975 report: &OrderStatusReport,
976 ts_now: UnixNanos,
977) -> OrderEventAny {
978 OrderEventAny::Canceled(OrderCanceled::new(
979 order.trader_id(),
980 order.strategy_id(),
981 order.instrument_id(),
982 order.client_order_id(),
983 UUID4::new(),
984 report.ts_last,
985 ts_now,
986 true, order.venue_order_id(),
988 order.account_id(),
989 report.cancel_reason.as_deref().map(Ustr::from),
990 ))
991}
992
993#[must_use]
995pub(super) fn create_reconciliation_expired(
996 order: &OrderAny,
997 report: &OrderStatusReport,
998 ts_now: UnixNanos,
999) -> OrderEventAny {
1000 OrderEventAny::Expired(OrderExpired::new(
1001 order.trader_id(),
1002 order.strategy_id(),
1003 order.instrument_id(),
1004 order.client_order_id(),
1005 UUID4::new(),
1006 report.ts_last,
1007 ts_now,
1008 true, order.venue_order_id(),
1010 order.account_id(),
1011 ))
1012}
1013
1014#[must_use]
1016pub(super) fn create_reconciliation_updated(
1017 order: &OrderAny,
1018 report: &OrderStatusReport,
1019 ts_now: UnixNanos,
1020) -> OrderEventAny {
1021 let trigger_price = match order.order_type() {
1028 OrderType::StopMarket
1029 | OrderType::StopLimit
1030 | OrderType::MarketIfTouched
1031 | OrderType::LimitIfTouched
1032 | OrderType::TrailingStopMarket
1033 | OrderType::TrailingStopLimit => report.trigger_price,
1034 _ => None,
1035 };
1036
1037 let price = if LIMIT_ORDER_TYPES.contains(&order.order_type())
1039 || order.order_type() == OrderType::MarketToLimit
1040 {
1041 report.price
1042 } else {
1043 None
1044 };
1045
1046 OrderEventAny::Updated(OrderUpdated::new(
1047 order.trader_id(),
1048 order.strategy_id(),
1049 order.instrument_id(),
1050 order.client_order_id(),
1051 report.quantity,
1052 UUID4::new(),
1053 report.ts_last,
1054 ts_now,
1055 true, order.venue_order_id(),
1057 order.account_id(),
1058 price,
1059 trigger_price,
1060 None, order.is_quote_quantity(),
1062 ))
1063}
1064
1065pub(super) fn create_inferred_fill(
1067 order: &OrderAny,
1068 report: &OrderStatusReport,
1069 account_id: AccountId,
1070 instrument: &InstrumentAny,
1071 ts_now: UnixNanos,
1072 commission: Option<Money>,
1073) -> Option<OrderEventAny> {
1074 let cached_order_side = order.order_side();
1075 let order_side = report.order_side.unwrap_or(cached_order_side);
1076 if order_side != cached_order_side {
1077 log::warn!(
1078 "Order side mismatch for {}: cached={:?}, venue={order_side:?}",
1079 order.client_order_id(),
1080 cached_order_side,
1081 );
1082 }
1083
1084 let liquidity_side = match order.order_type() {
1085 OrderType::Market
1086 | OrderType::StopMarket
1087 | OrderType::MarketToLimit
1088 | OrderType::TrailingStopMarket => LiquiditySide::Taker,
1089 _ if order.is_post_only() => LiquiditySide::Maker,
1090 _ => LiquiditySide::NoLiquiditySide,
1091 };
1092
1093 let Some(last_px) = resolve_fill_price(order, report, instrument) else {
1094 log::warn!(
1095 "Cannot create inferred fill for {}: no avg_px, report price, or order price",
1096 order.client_order_id()
1097 );
1098
1099 return None;
1100 };
1101 let last_px = clamp_inferred_fill_price(last_px, instrument);
1102
1103 let position_id = reconciliation_position_id(report, instrument);
1104 let trade_id = create_inferred_reconciliation_trade_id(
1105 account_id,
1106 order.instrument_id(),
1107 order.client_order_id(),
1108 Some(report.venue_order_id),
1109 order_side,
1110 order.order_type(),
1111 report.filled_qty,
1112 report.filled_qty,
1113 last_px,
1114 position_id,
1115 report.ts_last,
1116 );
1117
1118 log::info!(
1119 "Generated inferred fill for {} ({}) qty={} px={}",
1120 order.client_order_id(),
1121 report.venue_order_id,
1122 report.filled_qty,
1123 last_px,
1124 );
1125
1126 Some(OrderEventAny::Filled(OrderFilled::new(
1127 order.trader_id(),
1128 order.strategy_id(),
1129 order.instrument_id(),
1130 order.client_order_id(),
1131 report.venue_order_id,
1132 account_id,
1133 trade_id,
1134 order_side,
1135 order.order_type(),
1136 report.filled_qty,
1137 last_px,
1138 instrument.quote_currency(),
1139 liquidity_side,
1140 UUID4::new(),
1141 report.ts_last,
1142 ts_now,
1143 true, report.venue_position_id,
1145 commission,
1146 None,
1147 )))
1148}
1149
1150pub fn create_incremental_inferred_fill(
1152 order: &OrderAny,
1153 report: &OrderStatusReport,
1154 account_id: &AccountId,
1155 instrument: &InstrumentAny,
1156 ts_now: UnixNanos,
1157 commission: Option<Money>,
1158) -> Option<OrderEventAny> {
1159 let order_side = order.order_side();
1160
1161 let order_filled_qty = order.filled_qty();
1162 debug_assert!(
1163 report.filled_qty >= order_filled_qty,
1164 "incremental inferred fill requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
1165 report.filled_qty,
1166 order_filled_qty,
1167 order.client_order_id(),
1168 );
1169 let last_qty = report.filled_qty - order_filled_qty;
1170
1171 if last_qty <= Quantity::zero(instrument.size_precision()) {
1172 return None;
1173 }
1174
1175 let (last_px, liquidity_side) =
1176 incremental_inferred_fill_price_and_liquidity(order, report, instrument)?;
1177
1178 let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
1179 let position_id = reconciliation_position_id(report, instrument);
1180 let trade_id = create_inferred_reconciliation_trade_id(
1181 *account_id,
1182 order.instrument_id(),
1183 order.client_order_id(),
1184 Some(venue_order_id),
1185 order_side,
1186 order.order_type(),
1187 report.filled_qty,
1188 last_qty,
1189 last_px,
1190 position_id,
1191 report.ts_last,
1192 );
1193
1194 log::info!(
1195 color = LogColor::Blue as u8;
1196 "Generated inferred fill for {}: qty={}, px={}",
1197 order.client_order_id(),
1198 last_qty,
1199 last_px,
1200 );
1201
1202 Some(OrderEventAny::Filled(OrderFilled::new(
1203 order.trader_id(),
1204 order.strategy_id(),
1205 order.instrument_id(),
1206 order.client_order_id(),
1207 venue_order_id,
1208 *account_id,
1209 trade_id,
1210 order_side,
1211 order.order_type(),
1212 last_qty,
1213 last_px,
1214 instrument.quote_currency(),
1215 liquidity_side,
1216 UUID4::new(),
1217 report.ts_last,
1218 ts_now,
1219 true, report.venue_position_id,
1221 commission,
1222 None,
1223 )))
1224}
1225
1226#[must_use]
1234pub fn incremental_inferred_fill_price_and_liquidity(
1235 order: &OrderAny,
1236 report: &OrderStatusReport,
1237 instrument: &InstrumentAny,
1238) -> Option<(Price, LiquiditySide)> {
1239 let last_px = calculate_incremental_fill_price(order, report, instrument)?;
1240
1241 Some((
1242 clamp_inferred_fill_price(last_px, instrument),
1243 inferred_fill_liquidity_side(order),
1244 ))
1245}
1246
1247#[must_use]
1254pub fn inferred_fill_price_and_liquidity(
1255 order: &OrderAny,
1256 report: &OrderStatusReport,
1257 instrument: &InstrumentAny,
1258) -> Option<(Price, LiquiditySide)> {
1259 let last_px = resolve_fill_price(order, report, instrument)?;
1260
1261 Some((
1262 clamp_inferred_fill_price(last_px, instrument),
1263 inferred_fill_liquidity_side(order),
1264 ))
1265}
1266
1267fn inferred_fill_liquidity_side(order: &OrderAny) -> LiquiditySide {
1268 match order.order_type() {
1269 OrderType::Market
1270 | OrderType::StopMarket
1271 | OrderType::MarketToLimit
1272 | OrderType::TrailingStopMarket => LiquiditySide::Taker,
1273 _ if order.is_post_only() => LiquiditySide::Maker,
1274 _ => LiquiditySide::NoLiquiditySide,
1275 }
1276}
1277
1278pub fn create_inferred_fill_for_qty(
1284 order: &OrderAny,
1285 report: &OrderStatusReport,
1286 account_id: &AccountId,
1287 instrument: &InstrumentAny,
1288 fill_qty: Quantity,
1289 ts_now: UnixNanos,
1290 commission: Option<Money>,
1291) -> Option<OrderEventAny> {
1292 if fill_qty.is_zero() {
1293 return None;
1294 }
1295
1296 let order_side = order.order_side();
1297
1298 let Some((last_px, liquidity_side)) =
1299 inferred_fill_price_and_liquidity(order, report, instrument)
1300 else {
1301 log::warn!(
1302 "Cannot determine fill price for {}: no avg_px, report price, or order price",
1303 order.client_order_id()
1304 );
1305
1306 return None;
1307 };
1308
1309 let venue_order_id = order.venue_order_id().unwrap_or(report.venue_order_id);
1310 let position_id = reconciliation_position_id(report, instrument);
1311 let trade_id = create_inferred_reconciliation_trade_id(
1312 *account_id,
1313 order.instrument_id(),
1314 order.client_order_id(),
1315 Some(venue_order_id),
1316 order_side,
1317 order.order_type(),
1318 report.filled_qty,
1319 fill_qty,
1320 last_px,
1321 position_id,
1322 report.ts_last,
1323 );
1324
1325 log::info!(
1326 color = LogColor::Blue as u8;
1327 "Generated inferred fill for {}: qty={}, px={}",
1328 order.client_order_id(),
1329 fill_qty,
1330 last_px,
1331 );
1332
1333 Some(OrderEventAny::Filled(OrderFilled::new(
1334 order.trader_id(),
1335 order.strategy_id(),
1336 order.instrument_id(),
1337 order.client_order_id(),
1338 venue_order_id,
1339 *account_id,
1340 trade_id,
1341 order_side,
1342 order.order_type(),
1343 fill_qty,
1344 last_px,
1345 instrument.quote_currency(),
1346 liquidity_side,
1347 UUID4::new(),
1348 report.ts_last,
1349 ts_now,
1350 true, report.venue_position_id,
1352 commission,
1353 None,
1354 )))
1355}
1356
1357fn report_is_confirmed_state(report: &OrderStatusReport) -> bool {
1358 matches!(
1359 report.order_status,
1360 OrderStatus::Accepted
1361 | OrderStatus::Triggered
1362 | OrderStatus::PartiallyFilled
1363 | OrderStatus::Filled
1364 | OrderStatus::Canceled
1365 | OrderStatus::Expired
1366 )
1367}
1368
1369fn is_working(status: OrderStatus) -> bool {
1370 matches!(
1371 status,
1372 OrderStatus::Accepted | OrderStatus::Triggered | OrderStatus::PartiallyFilled
1373 )
1374}
1375
1376pub(crate) fn is_superseded_terminal_report(order: &OrderAny, report: &OrderStatusReport) -> bool {
1377 if !matches!(
1378 report.order_status,
1379 OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Filled
1380 ) {
1381 return false;
1382 }
1383
1384 let Some(cached_venue_order_id) = order.venue_order_id() else {
1385 return false;
1386 };
1387
1388 cached_venue_order_id != report.venue_order_id
1389 && order
1390 .venue_order_ids()
1391 .iter()
1392 .any(|venue_order_id| **venue_order_id == report.venue_order_id)
1393}
1394
1395fn pending_terminal_quantity_update(order: &OrderAny, report: &OrderStatusReport) -> bool {
1396 if !matches!(
1397 order.status(),
1398 OrderStatus::PendingUpdate | OrderStatus::PendingCancel
1399 ) {
1400 return false;
1401 }
1402
1403 if matches!(
1404 report.order_status,
1405 OrderStatus::Canceled | OrderStatus::Expired
1406 ) {
1407 return true;
1408 }
1409
1410 report.order_status == OrderStatus::Filled
1411 && report.filled_qty > order.quantity()
1412 && report.quantity >= report.filled_qty
1413}
1414
1415fn should_accept_before_reconciliation(order: &OrderAny, report: &OrderStatusReport) -> bool {
1416 order.status() == OrderStatus::Submitted && report.order_status != OrderStatus::Rejected
1417}
1418
1419fn reconcile_fill_quantity_mismatch(
1423 order: &OrderAny,
1424 report: &OrderStatusReport,
1425 instrument: Option<&InstrumentAny>,
1426 ts_now: UnixNanos,
1427 commission: Option<Money>,
1428) -> Option<OrderEventAny> {
1429 let order_filled_qty = order.filled_qty();
1430 let report_filled_qty = report.filled_qty;
1431
1432 if report_filled_qty < order_filled_qty {
1433 let precision = order_filled_qty.precision.max(report_filled_qty.precision);
1436 if is_within_single_unit_tolerance(
1437 report_filled_qty.as_decimal(),
1438 order_filled_qty.as_decimal(),
1439 precision,
1440 ) {
1441 return None;
1442 }
1443
1444 log::warn!(
1445 "Fill qty mismatch for {} ({}): cached={}, venue={}, order_qty={} (venue < cached)",
1446 order.client_order_id(),
1447 report.venue_order_id,
1448 order_filled_qty,
1449 report_filled_qty,
1450 order.quantity(),
1451 );
1452 return None;
1453 }
1454
1455 if report_filled_qty > order_filled_qty {
1456 if order.is_closed() {
1459 let precision = order_filled_qty.precision.max(report_filled_qty.precision);
1460
1461 if is_within_single_unit_tolerance(
1462 report_filled_qty.as_decimal(),
1463 order_filled_qty.as_decimal(),
1464 precision,
1465 ) {
1466 return None;
1467 }
1468
1469 log::debug!(
1470 "{} {} already closed but reported difference in filled_qty: \
1471 report={}, cached={}, skipping inferred fill generation for closed order",
1472 order.instrument_id(),
1473 order.client_order_id(),
1474 report_filled_qty,
1475 order_filled_qty,
1476 );
1477 return None;
1478 }
1479
1480 let Some(instrument) = instrument else {
1482 log::warn!(
1483 "Cannot generate inferred fill for {}: instrument not available",
1484 order.client_order_id()
1485 );
1486 return None;
1487 };
1488
1489 let account_id = order.account_id()?;
1490 return create_incremental_inferred_fill(
1491 order,
1492 report,
1493 &account_id,
1494 instrument,
1495 ts_now,
1496 commission,
1497 );
1498 }
1499
1500 if order.status() != report.order_status {
1505 if should_reconciliation_update(order, report) {
1506 log::info!(
1507 "Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, \
1508 filled_qty={}, updating quantity {}->{}",
1509 order.client_order_id(),
1510 order.status(),
1511 report.order_status,
1512 report.filled_qty,
1513 order.quantity(),
1514 report.quantity,
1515 );
1516 return Some(create_reconciliation_updated(order, report, ts_now));
1517 }
1518
1519 log::warn!(
1520 "Status mismatch with matching fill qty for {}: local={:?}, venue={:?}, filled_qty={}",
1521 order.client_order_id(),
1522 order.status(),
1523 report.order_status,
1524 report.filled_qty
1525 );
1526 }
1527
1528 None
1529}
1530
1531fn calculate_incremental_fill_price(
1537 order: &OrderAny,
1538 report: &OrderStatusReport,
1539 instrument: &InstrumentAny,
1540) -> Option<Price> {
1541 let order_filled_qty = order.filled_qty();
1542 debug_assert!(
1543 report.filled_qty >= order_filled_qty,
1544 "incremental fill price requires report.filled_qty ({}) >= order.filled_qty ({}) for {}",
1545 report.filled_qty,
1546 order_filled_qty,
1547 order.client_order_id(),
1548 );
1549
1550 if order_filled_qty.is_zero() {
1552 let last_px = resolve_fill_price(order, report, instrument);
1553 if last_px.is_none() {
1554 log::warn!(
1555 "Cannot determine fill price for {}: no avg_px, report price, or order price",
1556 order.client_order_id()
1557 );
1558 }
1559
1560 return last_px;
1561 }
1562
1563 if let Some(report_avg_px) = report.avg_px {
1565 let last_px_decimal = match order.avg_px() {
1566 None => report_avg_px,
1568 Some(order_avg_px) => {
1569 let report_filled_qty = report.filled_qty;
1570 let last_qty = report_filled_qty - order_filled_qty;
1571
1572 let report_notional = report_avg_px * report_filled_qty.as_decimal();
1573 let order_notional = order_avg_px * order_filled_qty.as_decimal();
1574 let last_notional = report_notional - order_notional;
1575 let back_solved = last_notional / last_qty.as_decimal();
1576
1577 if back_solved < Decimal::ZERO && !instrument.allows_negative_price() {
1578 if report_avg_px < Decimal::ZERO {
1579 log::warn!(
1580 "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",
1581 order.client_order_id(),
1582 );
1583
1584 return None;
1585 }
1586
1587 log::warn!(
1588 "Negative back-solved fill price {back_solved} for {}, using venue average {report_avg_px}",
1589 order.client_order_id(),
1590 );
1591
1592 report_avg_px
1593 } else {
1594 back_solved
1595 }
1596 }
1597 };
1598
1599 return reconciliation_price(last_px_decimal, instrument)
1600 .inspect_err(|e| {
1601 log::warn!(
1602 "Cannot price {} from incremental {last_px_decimal}, falling back: {e}",
1603 order.client_order_id(),
1604 );
1605 })
1606 .ok()
1607 .or_else(|| resolve_fill_price(order, report, instrument));
1608 }
1609
1610 resolve_fill_price(order, report, instrument)
1611}
1612
1613fn resolve_fill_price(
1619 order: &OrderAny,
1620 report: &OrderStatusReport,
1621 instrument: &InstrumentAny,
1622) -> Option<Price> {
1623 report
1624 .avg_px
1625 .and_then(|avg_px| {
1626 reconciliation_price(avg_px, instrument)
1627 .inspect_err(|e| {
1628 log::warn!(
1629 "Cannot price {} from venue average {avg_px}, trying next source: {e}",
1630 order.client_order_id(),
1631 );
1632 })
1633 .ok()
1634 })
1635 .or(report.price)
1636 .or_else(|| order.price())
1637}
1638
1639fn clamp_inferred_fill_price(price: Price, instrument: &InstrumentAny) -> Price {
1644 let px = cap_price_at_instrument_max(price.as_decimal(), instrument);
1645 Price::from_decimal_dp(px, price.precision.max(instrument.price_precision())).unwrap_or(price)
1646}
1647
1648fn reconciliation_position_id(
1649 report: &OrderStatusReport,
1650 instrument: &InstrumentAny,
1651) -> PositionId {
1652 report
1653 .venue_position_id
1654 .unwrap_or_else(|| PositionId::new(format!("{}-EXTERNAL", instrument.id())))
1655}