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.clear();
159 fill_map.clear();
160 order_map.insert(first_venue_order_id, order);
161 fill_map.insert(first_venue_order_id, vec![fill]);
162 }
163
164 FillAdjustmentResult::FilterToCurrentLifecycle {
165 last_zero_crossing_ts,
166 current_lifecycle_fills: _,
167 } => {
168 for fills in fill_map.values_mut() {
170 fills.retain(|f| f.ts_event.as_u64() > last_zero_crossing_ts);
171 }
172 fill_map.retain(|_, fills| !fills.is_empty());
173
174 let orders_with_fills: ahash::AHashSet<VenueOrderId> =
176 fill_map.keys().copied().collect();
177 order_map.retain(|id, order| {
178 orders_with_fills.contains(id)
179 || !matches!(
180 order.order_status,
181 OrderStatus::Denied
182 | OrderStatus::Rejected
183 | OrderStatus::Canceled
184 | OrderStatus::Expired
185 | OrderStatus::Filled
186 )
187 });
188 }
189
190 FillAdjustmentResult::AddSyntheticOpening { .. }
191 | FillAdjustmentResult::ReplaceCurrentLifecycle { .. } => {}
192 }
193
194 Ok(ReconciliationResult {
195 orders: order_map,
196 fills: fill_map,
197 })
198}
199
200#[must_use]
209pub(super) fn adjust_fills_for_partial_window(
210 fills: &[FillSnapshot],
211 venue_position: &VenuePositionSnapshot,
212 tolerance: Decimal,
213) -> FillAdjustmentResult {
214 if fills.is_empty() {
216 return FillAdjustmentResult::NoAdjustment;
217 }
218
219 if venue_position.qty == Decimal::ZERO {
221 return FillAdjustmentResult::NoAdjustment;
222 }
223
224 let zero_crossings = detect_zero_crossings(fills);
226
227 let venue_qty_signed = match venue_position.side {
229 PositionSide::Long => venue_position.qty,
230 PositionSide::Short => -venue_position.qty,
231 PositionSide::Flat => Decimal::ZERO,
232 };
233
234 if !zero_crossings.is_empty() {
236 let mut last_flat_crossing_ts = None;
239 let mut running_qty = Decimal::ZERO;
240
241 for fill in fills {
242 let prev_qty = running_qty;
243 running_qty += Decimal::from(fill.direction()) * fill.qty;
244
245 if prev_qty != Decimal::ZERO && running_qty == Decimal::ZERO {
246 last_flat_crossing_ts = Some(fill.ts_event);
247 }
248 }
249
250 let lifecycle_boundary_ts =
251 last_flat_crossing_ts.unwrap_or(*zero_crossings.last().unwrap());
252
253 let current_lifecycle_fills: Vec<FillSnapshot> = fills
255 .iter()
256 .filter(|f| f.ts_event > lifecycle_boundary_ts)
257 .cloned()
258 .collect();
259
260 if current_lifecycle_fills.is_empty() {
261 return FillAdjustmentResult::NoAdjustment;
262 }
263
264 let (current_qty, current_value) = simulate_position(¤t_lifecycle_fills);
266
267 if check_position_match(
269 current_qty,
270 current_value,
271 venue_qty_signed,
272 venue_position.avg_px,
273 tolerance,
274 ) {
275 return FillAdjustmentResult::FilterToCurrentLifecycle {
277 last_zero_crossing_ts: lifecycle_boundary_ts,
278 current_lifecycle_fills,
279 };
280 }
281
282 if let Some(first_fill) = current_lifecycle_fills.first() {
284 let synthetic_fill = FillSnapshot::new(
285 first_fill.venue_order_id,
286 position_to_order_side(venue_position.side),
287 venue_position.qty,
288 venue_position.avg_px,
289 first_fill.ts_event.saturating_sub(1), );
291
292 return FillAdjustmentResult::ReplaceCurrentLifecycle {
293 synthetic_fill,
294 first_venue_order_id: first_fill.venue_order_id,
295 };
296 }
297
298 return FillAdjustmentResult::NoAdjustment;
299 }
300
301 let oldest_lifecycle_fills: Vec<FillSnapshot> =
304 if let Some(&first_zero_crossing_ts) = zero_crossings.first() {
305 fills
307 .iter()
308 .filter(|f| f.ts_event <= first_zero_crossing_ts)
309 .cloned()
310 .collect()
311 } else {
312 fills.to_vec()
314 };
315
316 if oldest_lifecycle_fills.is_empty() {
317 return FillAdjustmentResult::NoAdjustment;
318 }
319
320 let (oldest_qty, oldest_value) = simulate_position(&oldest_lifecycle_fills);
322
323 if zero_crossings.is_empty() {
325 if check_position_match(
327 oldest_qty,
328 oldest_value,
329 venue_qty_signed,
330 venue_position.avg_px,
331 tolerance,
332 ) {
333 return FillAdjustmentResult::NoAdjustment;
334 }
335
336 if let Some(first_fill) = oldest_lifecycle_fills.first() {
338 let oldest_avg_px = if oldest_qty == Decimal::ZERO {
341 None
342 } else {
343 Some(oldest_value / oldest_qty.abs())
344 };
345
346 let reconciliation_price = calculate_reconciliation_price(
347 oldest_qty,
348 oldest_avg_px,
349 venue_qty_signed,
350 Some(venue_position.avg_px),
351 );
352
353 if let Some(opening_px) = reconciliation_price {
354 let opening_qty = if oldest_qty == Decimal::ZERO {
356 venue_qty_signed
357 } else {
358 venue_qty_signed - oldest_qty
360 };
361
362 if opening_qty.abs() > Decimal::ZERO {
363 let synthetic_side = if opening_qty > Decimal::ZERO {
364 OrderSide::Buy
365 } else {
366 OrderSide::Sell
367 };
368
369 let synthetic_fill = FillSnapshot::new(
370 first_fill.venue_order_id,
371 synthetic_side,
372 opening_qty.abs(),
373 opening_px,
374 first_fill.ts_event.saturating_sub(1),
375 );
376
377 return FillAdjustmentResult::AddSyntheticOpening {
378 synthetic_fill,
379 existing_fills: oldest_lifecycle_fills,
380 };
381 }
382 }
383 }
384
385 return FillAdjustmentResult::NoAdjustment;
386 }
387
388 if oldest_qty == Decimal::ZERO {
390 return FillAdjustmentResult::NoAdjustment;
392 }
393
394 if !oldest_lifecycle_fills.is_empty()
396 && let Some(&first_zero_crossing_ts) = zero_crossings.first()
397 {
398 let current_lifecycle_fills: Vec<FillSnapshot> = fills
400 .iter()
401 .filter(|f| f.ts_event > first_zero_crossing_ts)
402 .cloned()
403 .collect();
404
405 if !current_lifecycle_fills.is_empty()
406 && let Some(first_current_fill) = current_lifecycle_fills.first()
407 {
408 let synthetic_fill = FillSnapshot::new(
409 first_current_fill.venue_order_id,
410 position_to_order_side(venue_position.side),
411 venue_position.qty,
412 venue_position.avg_px,
413 first_current_fill.ts_event.saturating_sub(1),
414 );
415
416 return FillAdjustmentResult::AddSyntheticOpening {
417 synthetic_fill,
418 existing_fills: oldest_lifecycle_fills,
419 };
420 }
421 }
422
423 FillAdjustmentResult::NoAdjustment
424}
425
426pub fn check_position_reconciliation(
437 report: &PositionStatusReport,
438 cached_signed_qty: Decimal,
439 size_precision: Option<u8>,
440) -> bool {
441 let venue_signed_qty = report.signed_decimal_qty;
442
443 if venue_signed_qty == Decimal::ZERO && cached_signed_qty == Decimal::ZERO {
444 return true;
445 }
446
447 if let Some(precision) = size_precision
448 && is_within_single_unit_tolerance(cached_signed_qty, venue_signed_qty, precision)
449 {
450 log::debug!(
451 "Position for {} within tolerance: cached={}, venue={}",
452 report.instrument_id,
453 cached_signed_qty,
454 venue_signed_qty
455 );
456 return true;
457 }
458
459 if cached_signed_qty == venue_signed_qty {
460 return true;
461 }
462
463 log::warn!(
464 "Position discrepancy for {}: cached={}, venue={}",
465 report.instrument_id,
466 cached_signed_qty,
467 venue_signed_qty
468 );
469
470 false
471}
472
473pub(super) fn cap_price_at_instrument_max(px: Decimal, instrument: &InstrumentAny) -> Decimal {
491 let Some(max) = instrument.max_price() else {
492 return px;
493 };
494 let cap = max.as_decimal().round_dp_with_strategy(
495 u32::from(instrument.price_precision()),
496 RoundingStrategy::ToZero,
497 );
498 px.min(cap)
499}
500
501pub(super) fn create_synthetic_order_report(
511 fill: &FillSnapshot,
512 account_id: AccountId,
513 instrument_id: InstrumentId,
514 instrument: &InstrumentAny,
515 venue_order_id: VenueOrderId,
516) -> anyhow::Result<OrderStatusReport> {
517 let order_qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
518
519 let mut report = OrderStatusReport::new(
520 account_id,
521 instrument_id,
522 None, venue_order_id,
524 fill.side.into(),
525 OrderType::Market,
526 TimeInForce::Gtc,
527 OrderStatus::Filled,
528 order_qty,
529 order_qty, UnixNanos::from(fill.ts_event),
531 UnixNanos::from(fill.ts_event),
532 UnixNanos::from(fill.ts_event),
533 None, );
535 report.avg_px = Some(cap_price_at_instrument_max(fill.px, instrument));
536 Ok(report)
537}
538
539pub(super) fn create_synthetic_fill_report(
545 fill: &FillSnapshot,
546 account_id: AccountId,
547 instrument_id: InstrumentId,
548 instrument: &InstrumentAny,
549 venue_order_id: VenueOrderId,
550) -> anyhow::Result<FillReport> {
551 let trade_id = create_synthetic_trade_id(fill);
552 let qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
553 let px = Price::from_decimal_dp(
554 cap_price_at_instrument_max(fill.px, instrument),
555 instrument.price_precision(),
556 )?;
557
558 Ok(FillReport::new(
559 account_id,
560 instrument_id,
561 venue_order_id,
562 trade_id,
563 fill.side,
564 qty,
565 px,
566 Money::zero(instrument.quote_currency()),
567 LiquiditySide::NoLiquiditySide,
568 None, None, fill.ts_event.into(),
571 fill.ts_event.into(),
572 None, ))
574}
575
576fn position_report_to_snapshot(report: &PositionStatusReport) -> VenuePositionSnapshot {
578 VenuePositionSnapshot {
579 side: report.position_side,
580 qty: report.quantity.into(),
581 avg_px: report.avg_px_open.unwrap_or(Decimal::ZERO),
582 }
583}
584
585fn position_to_order_side(side: PositionSide) -> OrderSide {
591 match side {
592 PositionSide::Long => OrderSide::Buy,
593 PositionSide::Short => OrderSide::Sell,
594 PositionSide::Flat => {
595 unreachable!("flat venue position must be guarded by an earlier check")
596 }
597 }
598}
599
600fn extract_instrument_reports(
602 mass_status: &ExecutionMassStatus,
603 instrument_id: InstrumentId,
604) -> ReconciliationResult {
605 let mut orders = IndexMap::new();
606 let mut fills = IndexMap::new();
607
608 for (id, order) in mass_status.order_reports() {
609 if order.instrument_id == instrument_id {
610 orders.insert(id, order.clone());
611 }
612 }
613
614 for (id, fill_list) in mass_status.fill_reports() {
615 let filtered: Vec<_> = fill_list
616 .iter()
617 .filter(|f| f.instrument_id == instrument_id)
618 .cloned()
619 .collect();
620
621 if !filtered.is_empty() {
622 fills.insert(id, filtered);
623 }
624 }
625
626 ReconciliationResult { orders, fills }
627}
628
629struct ExtractedFills {
631 snapshots: Vec<FillSnapshot>,
632 orders: IndexMap<VenueOrderId, OrderStatusReport>,
633 fills: IndexMap<VenueOrderId, Vec<FillReport>>,
634}
635
636fn extract_fills_for_instrument(
638 mass_status: &ExecutionMassStatus,
639 instrument_id: InstrumentId,
640) -> ExtractedFills {
641 let mut snapshots = Vec::new();
642 let mut order_map = IndexMap::new();
643 let mut fill_map = IndexMap::new();
644
645 for (id, order) in mass_status.order_reports() {
647 if order.instrument_id == instrument_id {
648 order_map.insert(id, order.clone());
649 }
650 }
651
652 for (venue_order_id, fill_reports) in mass_status.fill_reports() {
654 for fill in fill_reports {
655 if fill.instrument_id == instrument_id {
656 let side = mass_status
657 .order_reports()
658 .get(&venue_order_id)
659 .and_then(|order| order.order_side)
660 .unwrap_or(fill.order_side);
661
662 snapshots.push(FillSnapshot::new(
663 venue_order_id,
664 side,
665 fill.last_qty.into(),
666 fill.last_px.into(),
667 fill.ts_event.as_u64(),
668 ));
669
670 fill_map
671 .entry(venue_order_id)
672 .or_insert_with(Vec::new)
673 .push(fill.clone());
674 }
675 }
676 }
677
678 snapshots.sort_by_key(|f| f.ts_event);
680
681 ExtractedFills {
682 snapshots,
683 orders: order_map,
684 fills: fill_map,
685 }
686}
687
688#[must_use]
694pub(super) fn simulate_position(fills: &[FillSnapshot]) -> (Decimal, Decimal) {
695 let mut qty = Decimal::ZERO;
696 let mut value = Decimal::ZERO;
697
698 for fill in fills {
699 debug_assert!(
700 fill.qty > Decimal::ZERO,
701 "fill snapshot qty must be positive, received {}",
702 fill.qty,
703 );
704 let direction = Decimal::from(fill.direction());
705 let new_qty = qty + (direction * fill.qty);
706
707 if (qty >= Decimal::ZERO && direction > Decimal::ZERO)
709 || (qty <= Decimal::ZERO && direction < Decimal::ZERO)
710 {
711 value += fill.qty * fill.px;
713 qty = new_qty;
714 } else {
715 if qty.abs() >= fill.qty {
717 let close_ratio = fill.qty / qty.abs();
719 value *= Decimal::ONE - close_ratio;
720 qty = new_qty;
721 } else {
722 let remaining = fill.qty - qty.abs();
724 qty = direction * remaining;
725 value = remaining * fill.px;
726 }
727 }
728 }
729
730 debug_assert!(
731 value >= Decimal::ZERO,
732 "simulated position value must be non-negative, was {value}",
733 );
734 debug_assert!(
735 !(qty != Decimal::ZERO && value.is_sign_negative()),
736 "simulated avg price invariant: qty={qty}, value={value}",
737 );
738
739 (qty, value)
740}
741
742#[must_use]
751pub(super) fn detect_zero_crossings(fills: &[FillSnapshot]) -> Vec<u64> {
752 let mut running_qty = Decimal::ZERO;
753 let mut zero_crossings = Vec::new();
754
755 for fill in fills {
756 let prev_qty = running_qty;
757 running_qty += Decimal::from(fill.direction()) * fill.qty;
758
759 if prev_qty != Decimal::ZERO {
761 if running_qty == Decimal::ZERO {
762 zero_crossings.push(fill.ts_event);
764 } else if (prev_qty > Decimal::ZERO) != (running_qty > Decimal::ZERO) {
765 zero_crossings.push(fill.ts_event);
767 }
768 }
769 }
770
771 zero_crossings
772}
773
774#[must_use]
780pub(super) fn check_position_match(
781 simulated_qty: Decimal,
782 simulated_value: Decimal,
783 venue_qty: Decimal,
784 venue_avg_px: Decimal,
785 tolerance: Decimal,
786) -> bool {
787 if simulated_qty != venue_qty {
788 return false;
789 }
790
791 if simulated_qty == Decimal::ZERO {
792 return true; }
794
795 let abs_qty = simulated_qty.abs();
797 if abs_qty == Decimal::ZERO {
798 return false;
799 }
800
801 let simulated_avg_px = simulated_value / abs_qty;
802
803 if venue_avg_px == Decimal::ZERO {
805 return false;
806 }
807
808 let relative_diff = (simulated_avg_px - venue_avg_px).abs() / venue_avg_px.abs();
809
810 relative_diff <= tolerance
811}
812
813pub fn calculate_reconciliation_price(
831 current_position_qty: Decimal,
832 current_position_avg_px: Option<Decimal>,
833 target_position_qty: Decimal,
834 target_position_avg_px: Option<Decimal>,
835) -> Option<Decimal> {
836 let qty_diff = target_position_qty - current_position_qty;
837
838 if qty_diff == Decimal::ZERO {
839 return None; }
841
842 if target_position_qty == Decimal::ZERO {
845 return current_position_avg_px;
846 }
847
848 let target_avg_px = target_position_avg_px?;
850 if target_avg_px == Decimal::ZERO {
851 return None;
852 }
853
854 if current_position_qty == Decimal::ZERO || current_position_avg_px.is_none() {
856 return Some(target_avg_px);
857 }
858
859 let current_avg_px = current_position_avg_px?;
860
861 let is_flip = (current_position_qty > Decimal::ZERO) != (target_position_qty > Decimal::ZERO)
864 && target_position_qty != Decimal::ZERO;
865
866 if is_flip {
867 return Some(target_avg_px);
868 }
869
870 let target_value = target_position_qty * target_avg_px;
873 let current_value = current_position_qty * current_avg_px;
874 let diff_value = target_value - current_value;
875
876 let reconciliation_px = diff_value / qty_diff;
878
879 if reconciliation_px > Decimal::ZERO {
881 return Some(reconciliation_px);
882 }
883
884 None
885}
886
887#[must_use]
892pub(super) fn is_within_single_unit_tolerance(
893 value1: Decimal,
894 value2: Decimal,
895 precision: u8,
896) -> bool {
897 if precision == 0 {
898 return value1 == value2;
899 }
900
901 let tolerance = Decimal::new(1, u32::from(precision));
902 let difference = (value1 - value2).abs();
903 difference <= tolerance
904}