1use indexmap::IndexMap;
24use nautilus_core::UnixNanos;
25use nautilus_model::{
26 enums::{LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce},
27 identifiers::{AccountId, InstrumentId, VenueOrderId},
28 instruments::{Instrument, InstrumentAny},
29 reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
30 types::{Money, Price, Quantity},
31};
32use rust_decimal::{Decimal, RoundingStrategy};
33
34use super::{
35 ids::{create_synthetic_trade_id, create_synthetic_venue_order_id},
36 types::{FillAdjustmentResult, FillSnapshot, ReconciliationResult, VenuePositionSnapshot},
37};
38
39const DEFAULT_TOLERANCE: Decimal = Decimal::from_parts(1, 0, 0, false, 4); pub fn process_mass_status_for_reconciliation(
52 mass_status: &ExecutionMassStatus,
53 instrument: &InstrumentAny,
54 tolerance: Option<Decimal>,
55) -> anyhow::Result<ReconciliationResult> {
56 process_mass_status_for_reconciliation_inner(mass_status, instrument, tolerance, true)
57}
58
59pub fn process_mass_status_for_reconciliation_without_synthetic_reports(
67 mass_status: &ExecutionMassStatus,
68 instrument: &InstrumentAny,
69 tolerance: Option<Decimal>,
70) -> anyhow::Result<ReconciliationResult> {
71 process_mass_status_for_reconciliation_inner(mass_status, instrument, tolerance, false)
72}
73
74fn process_mass_status_for_reconciliation_inner(
75 mass_status: &ExecutionMassStatus,
76 instrument: &InstrumentAny,
77 tolerance: Option<Decimal>,
78 generate_synthetic_reports: bool,
79) -> anyhow::Result<ReconciliationResult> {
80 let instrument_id = instrument.id();
81 let account_id = mass_status.account_id;
82 let tol = tolerance.unwrap_or(DEFAULT_TOLERANCE);
83
84 let position_reports = mass_status.position_reports();
86 let venue_position = match position_reports.get(&instrument_id).and_then(|r| r.first()) {
87 Some(report) => position_report_to_snapshot(report),
88 None => {
89 return Ok(extract_instrument_reports(mass_status, instrument_id));
91 }
92 };
93
94 let extracted = extract_fills_for_instrument(mass_status, instrument_id);
96 let fill_snapshots = extracted.snapshots;
97 let mut order_map = extracted.orders;
98 let mut fill_map = extracted.fills;
99
100 if fill_snapshots.is_empty() {
101 return Ok(ReconciliationResult {
102 orders: order_map,
103 fills: fill_map,
104 });
105 }
106
107 let result = adjust_fills_for_partial_window(&fill_snapshots, &venue_position, tol);
109
110 match result {
112 FillAdjustmentResult::NoAdjustment => {}
113
114 FillAdjustmentResult::AddSyntheticOpening {
115 synthetic_fill,
116 existing_fills: _,
117 } if generate_synthetic_reports => {
118 let venue_order_id = create_synthetic_venue_order_id(&synthetic_fill, instrument_id);
119 let order = create_synthetic_order_report(
120 &synthetic_fill,
121 account_id,
122 instrument_id,
123 instrument,
124 venue_order_id,
125 )?;
126 let fill = create_synthetic_fill_report(
127 &synthetic_fill,
128 account_id,
129 instrument_id,
130 instrument,
131 venue_order_id,
132 )?;
133
134 order_map.insert(venue_order_id, order);
135 fill_map.entry(venue_order_id).or_default().insert(0, fill);
136 }
137
138 FillAdjustmentResult::ReplaceCurrentLifecycle {
139 synthetic_fill,
140 first_venue_order_id,
141 } if generate_synthetic_reports => {
142 let order = create_synthetic_order_report(
143 &synthetic_fill,
144 account_id,
145 instrument_id,
146 instrument,
147 first_venue_order_id,
148 )?;
149 let fill = create_synthetic_fill_report(
150 &synthetic_fill,
151 account_id,
152 instrument_id,
153 instrument,
154 first_venue_order_id,
155 )?;
156
157 order_map.retain(|_, order| {
159 order.filled_qty.is_zero()
160 && (order.order_status.is_open() || order.order_status.is_inflight())
161 });
162 fill_map.clear();
163 order_map.insert(first_venue_order_id, order);
164 fill_map.insert(first_venue_order_id, vec![fill]);
165 }
166
167 FillAdjustmentResult::FilterToCurrentLifecycle {
168 last_zero_crossing_ts,
169 current_lifecycle_fills: _,
170 } => {
171 for fills in fill_map.values_mut() {
173 fills.retain(|f| f.ts_event.as_u64() > last_zero_crossing_ts);
174 }
175 fill_map.retain(|_, fills| !fills.is_empty());
176
177 let orders_with_fills: ahash::AHashSet<VenueOrderId> =
179 fill_map.keys().copied().collect();
180 order_map.retain(|id, order| {
181 orders_with_fills.contains(id)
182 || !matches!(
183 order.order_status,
184 OrderStatus::Denied
185 | OrderStatus::Rejected
186 | OrderStatus::Canceled
187 | OrderStatus::Expired
188 | OrderStatus::Filled
189 )
190 });
191 }
192
193 FillAdjustmentResult::AddSyntheticOpening { .. }
194 | FillAdjustmentResult::ReplaceCurrentLifecycle { .. } => {}
195 }
196
197 Ok(ReconciliationResult {
198 orders: order_map,
199 fills: fill_map,
200 })
201}
202
203#[must_use]
212pub(super) fn adjust_fills_for_partial_window(
213 fills: &[FillSnapshot],
214 venue_position: &VenuePositionSnapshot,
215 tolerance: Decimal,
216) -> FillAdjustmentResult {
217 if fills.is_empty() {
219 return FillAdjustmentResult::NoAdjustment;
220 }
221
222 if venue_position.qty == Decimal::ZERO {
224 return FillAdjustmentResult::NoAdjustment;
225 }
226
227 let zero_crossings = detect_zero_crossings(fills);
229
230 let venue_qty_signed = match venue_position.side {
232 PositionSide::Long => venue_position.qty,
233 PositionSide::Short => -venue_position.qty,
234 PositionSide::Flat => Decimal::ZERO,
235 };
236
237 if !zero_crossings.is_empty() {
239 let mut last_flat_crossing_ts = None;
242 let mut running_qty = Decimal::ZERO;
243
244 for fill in fills {
245 let prev_qty = running_qty;
246 running_qty += Decimal::from(fill.direction()) * fill.qty;
247
248 if prev_qty != Decimal::ZERO && running_qty == Decimal::ZERO {
249 last_flat_crossing_ts = Some(fill.ts_event);
250 }
251 }
252
253 let lifecycle_boundary_ts =
254 last_flat_crossing_ts.unwrap_or(*zero_crossings.last().unwrap());
255
256 let current_lifecycle_fills: Vec<FillSnapshot> = fills
258 .iter()
259 .filter(|f| f.ts_event > lifecycle_boundary_ts)
260 .cloned()
261 .collect();
262
263 if current_lifecycle_fills.is_empty() {
264 return FillAdjustmentResult::NoAdjustment;
265 }
266
267 let (current_qty, current_value) = simulate_position(¤t_lifecycle_fills);
269
270 if check_position_match(
272 current_qty,
273 current_value,
274 venue_qty_signed,
275 venue_position.avg_px,
276 tolerance,
277 ) {
278 return FillAdjustmentResult::FilterToCurrentLifecycle {
280 last_zero_crossing_ts: lifecycle_boundary_ts,
281 current_lifecycle_fills,
282 };
283 }
284
285 if let Some(first_fill) = current_lifecycle_fills.first() {
287 let synthetic_fill = FillSnapshot::new(
288 first_fill.venue_order_id,
289 position_to_order_side(venue_position.side),
290 venue_position.qty,
291 venue_position.avg_px,
292 first_fill.ts_event.saturating_sub(1), );
294
295 return FillAdjustmentResult::ReplaceCurrentLifecycle {
296 synthetic_fill,
297 first_venue_order_id: first_fill.venue_order_id,
298 };
299 }
300
301 return FillAdjustmentResult::NoAdjustment;
302 }
303
304 let oldest_lifecycle_fills: Vec<FillSnapshot> =
307 if let Some(&first_zero_crossing_ts) = zero_crossings.first() {
308 fills
310 .iter()
311 .filter(|f| f.ts_event <= first_zero_crossing_ts)
312 .cloned()
313 .collect()
314 } else {
315 fills.to_vec()
317 };
318
319 if oldest_lifecycle_fills.is_empty() {
320 return FillAdjustmentResult::NoAdjustment;
321 }
322
323 let (oldest_qty, oldest_value) = simulate_position(&oldest_lifecycle_fills);
325
326 if zero_crossings.is_empty() {
328 if check_position_match(
330 oldest_qty,
331 oldest_value,
332 venue_qty_signed,
333 venue_position.avg_px,
334 tolerance,
335 ) {
336 return FillAdjustmentResult::NoAdjustment;
337 }
338
339 if let Some(first_fill) = oldest_lifecycle_fills.first() {
341 let oldest_avg_px = if oldest_qty == Decimal::ZERO {
344 None
345 } else {
346 Some(oldest_value / oldest_qty.abs())
347 };
348
349 let reconciliation_price = calculate_reconciliation_price(
350 oldest_qty,
351 oldest_avg_px,
352 venue_qty_signed,
353 Some(venue_position.avg_px),
354 );
355
356 if let Some(opening_px) = reconciliation_price {
357 let opening_qty = if oldest_qty == Decimal::ZERO {
359 venue_qty_signed
360 } else {
361 venue_qty_signed - oldest_qty
363 };
364
365 if opening_qty.abs() > Decimal::ZERO {
366 let synthetic_side = if opening_qty > Decimal::ZERO {
367 OrderSide::Buy
368 } else {
369 OrderSide::Sell
370 };
371
372 let synthetic_fill = FillSnapshot::new(
373 first_fill.venue_order_id,
374 synthetic_side,
375 opening_qty.abs(),
376 opening_px,
377 first_fill.ts_event.saturating_sub(1),
378 );
379
380 return FillAdjustmentResult::AddSyntheticOpening {
381 synthetic_fill,
382 existing_fills: oldest_lifecycle_fills,
383 };
384 }
385 }
386 }
387
388 return FillAdjustmentResult::NoAdjustment;
389 }
390
391 if oldest_qty == Decimal::ZERO {
393 return FillAdjustmentResult::NoAdjustment;
395 }
396
397 if !oldest_lifecycle_fills.is_empty()
399 && let Some(&first_zero_crossing_ts) = zero_crossings.first()
400 {
401 let current_lifecycle_fills: Vec<FillSnapshot> = fills
403 .iter()
404 .filter(|f| f.ts_event > first_zero_crossing_ts)
405 .cloned()
406 .collect();
407
408 if !current_lifecycle_fills.is_empty()
409 && let Some(first_current_fill) = current_lifecycle_fills.first()
410 {
411 let synthetic_fill = FillSnapshot::new(
412 first_current_fill.venue_order_id,
413 position_to_order_side(venue_position.side),
414 venue_position.qty,
415 venue_position.avg_px,
416 first_current_fill.ts_event.saturating_sub(1),
417 );
418
419 return FillAdjustmentResult::AddSyntheticOpening {
420 synthetic_fill,
421 existing_fills: oldest_lifecycle_fills,
422 };
423 }
424 }
425
426 FillAdjustmentResult::NoAdjustment
427}
428
429pub fn check_position_reconciliation(
440 report: &PositionStatusReport,
441 cached_signed_qty: Decimal,
442 size_precision: Option<u8>,
443) -> bool {
444 let venue_signed_qty = report.signed_decimal_qty;
445
446 if venue_signed_qty == Decimal::ZERO && cached_signed_qty == Decimal::ZERO {
447 return true;
448 }
449
450 if let Some(precision) = size_precision
451 && is_within_single_unit_tolerance(cached_signed_qty, venue_signed_qty, precision)
452 {
453 log::debug!(
454 "Position for {} within tolerance: cached={}, venue={}",
455 report.instrument_id,
456 cached_signed_qty,
457 venue_signed_qty
458 );
459 return true;
460 }
461
462 if cached_signed_qty == venue_signed_qty {
463 return true;
464 }
465
466 log::warn!(
467 "Position discrepancy for {}: cached={}, venue={}",
468 report.instrument_id,
469 cached_signed_qty,
470 venue_signed_qty
471 );
472
473 false
474}
475
476pub(super) fn cap_price_at_instrument_max(px: Decimal, instrument: &InstrumentAny) -> Decimal {
494 let Some(max) = instrument.max_price() else {
495 return px;
496 };
497 let cap = max.as_decimal().round_dp_with_strategy(
498 u32::from(instrument.price_precision()),
499 RoundingStrategy::ToZero,
500 );
501 px.min(cap)
502}
503
504pub(super) fn create_synthetic_order_report(
514 fill: &FillSnapshot,
515 account_id: AccountId,
516 instrument_id: InstrumentId,
517 instrument: &InstrumentAny,
518 venue_order_id: VenueOrderId,
519) -> anyhow::Result<OrderStatusReport> {
520 let order_qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
521
522 let mut report = OrderStatusReport::new(
523 account_id,
524 instrument_id,
525 None, venue_order_id,
527 fill.side.into(),
528 OrderType::Market,
529 TimeInForce::Gtc,
530 OrderStatus::Filled,
531 order_qty,
532 order_qty, UnixNanos::from(fill.ts_event),
534 UnixNanos::from(fill.ts_event),
535 UnixNanos::from(fill.ts_event),
536 None, );
538 report.avg_px = Some(cap_price_at_instrument_max(fill.px, instrument));
539 Ok(report)
540}
541
542pub(super) fn create_synthetic_fill_report(
548 fill: &FillSnapshot,
549 account_id: AccountId,
550 instrument_id: InstrumentId,
551 instrument: &InstrumentAny,
552 venue_order_id: VenueOrderId,
553) -> anyhow::Result<FillReport> {
554 let trade_id = create_synthetic_trade_id(fill);
555 let qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
556 let px = Price::from_decimal_dp(
557 cap_price_at_instrument_max(fill.px, instrument),
558 instrument.price_precision(),
559 )?;
560
561 Ok(FillReport::new(
562 account_id,
563 instrument_id,
564 venue_order_id,
565 trade_id,
566 fill.side,
567 qty,
568 px,
569 Money::zero(instrument.quote_currency()),
570 LiquiditySide::NoLiquiditySide,
571 None, None, fill.ts_event.into(),
574 fill.ts_event.into(),
575 None, ))
577}
578
579fn position_report_to_snapshot(report: &PositionStatusReport) -> VenuePositionSnapshot {
581 VenuePositionSnapshot {
582 side: report.position_side,
583 qty: report.quantity.into(),
584 avg_px: report.avg_px_open.unwrap_or(Decimal::ZERO),
585 }
586}
587
588fn position_to_order_side(side: PositionSide) -> OrderSide {
594 match side {
595 PositionSide::Long => OrderSide::Buy,
596 PositionSide::Short => OrderSide::Sell,
597 PositionSide::Flat => {
598 unreachable!("flat venue position must be guarded by an earlier check")
599 }
600 }
601}
602
603fn extract_instrument_reports(
605 mass_status: &ExecutionMassStatus,
606 instrument_id: InstrumentId,
607) -> ReconciliationResult {
608 let mut orders = IndexMap::new();
609 let mut fills = IndexMap::new();
610
611 for (id, order) in mass_status.order_reports() {
612 if order.instrument_id == instrument_id {
613 orders.insert(id, order.clone());
614 }
615 }
616
617 for (id, fill_list) in mass_status.fill_reports() {
618 let filtered: Vec<_> = fill_list
619 .iter()
620 .filter(|f| {
621 if f.instrument_id != instrument_id {
622 return false;
623 }
624
625 if f.last_qty.is_zero() {
626 log::warn!("Skipping zero-quantity fill report: {f}");
627 return false;
628 }
629
630 true
631 })
632 .cloned()
633 .collect();
634
635 if !filtered.is_empty() {
636 fills.insert(id, filtered);
637 }
638 }
639
640 ReconciliationResult { orders, fills }
641}
642
643struct ExtractedFills {
645 snapshots: Vec<FillSnapshot>,
646 orders: IndexMap<VenueOrderId, OrderStatusReport>,
647 fills: IndexMap<VenueOrderId, Vec<FillReport>>,
648}
649
650fn extract_fills_for_instrument(
652 mass_status: &ExecutionMassStatus,
653 instrument_id: InstrumentId,
654) -> ExtractedFills {
655 let mut snapshots = Vec::new();
656 let mut order_map = IndexMap::new();
657 let mut fill_map = IndexMap::new();
658
659 for (id, order) in mass_status.order_reports() {
661 if order.instrument_id == instrument_id {
662 order_map.insert(id, order.clone());
663 }
664 }
665
666 for (venue_order_id, fill_reports) in mass_status.fill_reports() {
668 for fill in fill_reports {
669 if fill.instrument_id == instrument_id {
670 if fill.last_qty.is_zero() {
671 log::warn!("Skipping zero-quantity fill report: {fill}");
672 continue;
673 }
674
675 let side = mass_status
676 .order_reports()
677 .get(&venue_order_id)
678 .and_then(|order| order.order_side)
679 .unwrap_or(fill.order_side);
680
681 snapshots.push(FillSnapshot::new(
682 venue_order_id,
683 side,
684 fill.last_qty.into(),
685 fill.last_px.into(),
686 fill.ts_event.as_u64(),
687 ));
688
689 fill_map
690 .entry(venue_order_id)
691 .or_insert_with(Vec::new)
692 .push(fill.clone());
693 }
694 }
695 }
696
697 snapshots.sort_by_key(|f| f.ts_event);
699
700 ExtractedFills {
701 snapshots,
702 orders: order_map,
703 fills: fill_map,
704 }
705}
706
707#[must_use]
713pub(super) fn simulate_position(fills: &[FillSnapshot]) -> (Decimal, Decimal) {
714 let mut qty = Decimal::ZERO;
715 let mut value = Decimal::ZERO;
716
717 for fill in fills {
718 debug_assert!(
719 fill.qty > Decimal::ZERO,
720 "fill snapshot qty must be positive, received {}",
721 fill.qty,
722 );
723 let direction = Decimal::from(fill.direction());
724 let new_qty = qty + (direction * fill.qty);
725
726 if (qty >= Decimal::ZERO && direction > Decimal::ZERO)
728 || (qty <= Decimal::ZERO && direction < Decimal::ZERO)
729 {
730 value += fill.qty * fill.px;
732 qty = new_qty;
733 } else {
734 if qty.abs() >= fill.qty {
736 let close_ratio = fill.qty / qty.abs();
738 value *= Decimal::ONE - close_ratio;
739 qty = new_qty;
740 } else {
741 let remaining = fill.qty - qty.abs();
743 qty = direction * remaining;
744 value = remaining * fill.px;
745 }
746 }
747 }
748
749 debug_assert!(
750 value >= Decimal::ZERO,
751 "simulated position value must be non-negative, was {value}",
752 );
753 debug_assert!(
754 !(qty != Decimal::ZERO && value.is_sign_negative()),
755 "simulated avg price invariant: qty={qty}, value={value}",
756 );
757
758 (qty, value)
759}
760
761#[must_use]
770pub(super) fn detect_zero_crossings(fills: &[FillSnapshot]) -> Vec<u64> {
771 let mut running_qty = Decimal::ZERO;
772 let mut zero_crossings = Vec::new();
773
774 for fill in fills {
775 let prev_qty = running_qty;
776 running_qty += Decimal::from(fill.direction()) * fill.qty;
777
778 if prev_qty != Decimal::ZERO {
780 if running_qty == Decimal::ZERO {
781 zero_crossings.push(fill.ts_event);
783 } else if (prev_qty > Decimal::ZERO) != (running_qty > Decimal::ZERO) {
784 zero_crossings.push(fill.ts_event);
786 }
787 }
788 }
789
790 zero_crossings
791}
792
793#[must_use]
799pub(super) fn check_position_match(
800 simulated_qty: Decimal,
801 simulated_value: Decimal,
802 venue_qty: Decimal,
803 venue_avg_px: Decimal,
804 tolerance: Decimal,
805) -> bool {
806 if simulated_qty != venue_qty {
807 return false;
808 }
809
810 if simulated_qty == Decimal::ZERO {
811 return true; }
813
814 let abs_qty = simulated_qty.abs();
816 if abs_qty == Decimal::ZERO {
817 return false;
818 }
819
820 let simulated_avg_px = simulated_value / abs_qty;
821
822 if venue_avg_px == Decimal::ZERO {
824 return false;
825 }
826
827 let relative_diff = (simulated_avg_px - venue_avg_px).abs() / venue_avg_px.abs();
828
829 relative_diff <= tolerance
830}
831
832pub fn calculate_reconciliation_price(
850 current_position_qty: Decimal,
851 current_position_avg_px: Option<Decimal>,
852 target_position_qty: Decimal,
853 target_position_avg_px: Option<Decimal>,
854) -> Option<Decimal> {
855 let qty_diff = target_position_qty - current_position_qty;
856
857 if qty_diff == Decimal::ZERO {
858 return None; }
860
861 if target_position_qty == Decimal::ZERO {
864 return current_position_avg_px;
865 }
866
867 let target_avg_px = target_position_avg_px?;
869 if target_avg_px == Decimal::ZERO {
870 return None;
871 }
872
873 if current_position_qty == Decimal::ZERO || current_position_avg_px.is_none() {
875 return Some(target_avg_px);
876 }
877
878 let current_avg_px = current_position_avg_px?;
879
880 let is_flip = (current_position_qty > Decimal::ZERO) != (target_position_qty > Decimal::ZERO)
883 && target_position_qty != Decimal::ZERO;
884
885 if is_flip {
886 return Some(target_avg_px);
887 }
888
889 let target_value = target_position_qty * target_avg_px;
892 let current_value = current_position_qty * current_avg_px;
893 let diff_value = target_value - current_value;
894
895 let reconciliation_px = diff_value / qty_diff;
897
898 if reconciliation_px > Decimal::ZERO {
900 return Some(reconciliation_px);
901 }
902
903 None
904}
905
906#[must_use]
911pub(super) fn is_within_single_unit_tolerance(
912 value1: Decimal,
913 value2: Decimal,
914 precision: u8,
915) -> bool {
916 if precision == 0 {
917 return value1 == value2;
918 }
919
920 let tolerance = Decimal::new(1, u32::from(precision));
921 let difference = (value1 - value2).abs();
922 difference <= tolerance
923}