1use indexmap::IndexMap;
24use nautilus_core::UnixNanos;
25use nautilus_model::{
26 enums::{LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSideSpecified, 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 let instrument_id = instrument.id();
57 let account_id = mass_status.account_id;
58 let tol = tolerance.unwrap_or(DEFAULT_TOLERANCE);
59
60 let position_reports = mass_status.position_reports();
62 let venue_position = match position_reports.get(&instrument_id).and_then(|r| r.first()) {
63 Some(report) => position_report_to_snapshot(report),
64 None => {
65 return Ok(extract_instrument_reports(mass_status, instrument_id));
67 }
68 };
69
70 let extracted = extract_fills_for_instrument(mass_status, instrument_id);
72 let fill_snapshots = extracted.snapshots;
73 let mut order_map = extracted.orders;
74 let mut fill_map = extracted.fills;
75
76 if fill_snapshots.is_empty() {
77 return Ok(ReconciliationResult {
78 orders: order_map,
79 fills: fill_map,
80 });
81 }
82
83 let result = adjust_fills_for_partial_window(&fill_snapshots, &venue_position, tol);
85
86 match result {
88 FillAdjustmentResult::NoAdjustment => {}
89
90 FillAdjustmentResult::AddSyntheticOpening {
91 synthetic_fill,
92 existing_fills: _,
93 } => {
94 let venue_order_id = create_synthetic_venue_order_id(&synthetic_fill, instrument_id);
95 let order = create_synthetic_order_report(
96 &synthetic_fill,
97 account_id,
98 instrument_id,
99 instrument,
100 venue_order_id,
101 )?;
102 let fill = create_synthetic_fill_report(
103 &synthetic_fill,
104 account_id,
105 instrument_id,
106 instrument,
107 venue_order_id,
108 )?;
109
110 order_map.insert(venue_order_id, order);
111 fill_map.entry(venue_order_id).or_default().insert(0, fill);
112 }
113
114 FillAdjustmentResult::ReplaceCurrentLifecycle {
115 synthetic_fill,
116 first_venue_order_id,
117 } => {
118 let order = create_synthetic_order_report(
119 &synthetic_fill,
120 account_id,
121 instrument_id,
122 instrument,
123 first_venue_order_id,
124 )?;
125 let fill = create_synthetic_fill_report(
126 &synthetic_fill,
127 account_id,
128 instrument_id,
129 instrument,
130 first_venue_order_id,
131 )?;
132
133 order_map.clear();
135 fill_map.clear();
136 order_map.insert(first_venue_order_id, order);
137 fill_map.insert(first_venue_order_id, vec![fill]);
138 }
139
140 FillAdjustmentResult::FilterToCurrentLifecycle {
141 last_zero_crossing_ts,
142 current_lifecycle_fills: _,
143 } => {
144 for fills in fill_map.values_mut() {
146 fills.retain(|f| f.ts_event.as_u64() > last_zero_crossing_ts);
147 }
148 fill_map.retain(|_, fills| !fills.is_empty());
149
150 let orders_with_fills: ahash::AHashSet<VenueOrderId> =
152 fill_map.keys().copied().collect();
153 order_map.retain(|id, order| {
154 orders_with_fills.contains(id)
155 || !matches!(
156 order.order_status,
157 OrderStatus::Denied
158 | OrderStatus::Rejected
159 | OrderStatus::Canceled
160 | OrderStatus::Expired
161 | OrderStatus::Filled
162 )
163 });
164 }
165 }
166
167 Ok(ReconciliationResult {
168 orders: order_map,
169 fills: fill_map,
170 })
171}
172
173#[must_use]
183pub(super) fn adjust_fills_for_partial_window(
184 fills: &[FillSnapshot],
185 venue_position: &VenuePositionSnapshot,
186 tolerance: Decimal,
187) -> FillAdjustmentResult {
188 if fills.is_empty() {
190 return FillAdjustmentResult::NoAdjustment;
191 }
192
193 if venue_position.qty == Decimal::ZERO {
195 return FillAdjustmentResult::NoAdjustment;
196 }
197
198 let zero_crossings = detect_zero_crossings(fills);
200
201 let venue_qty_signed = match venue_position.side {
203 PositionSideSpecified::Long => venue_position.qty,
204 PositionSideSpecified::Short => -venue_position.qty,
205 PositionSideSpecified::Flat => Decimal::ZERO,
206 };
207
208 if !zero_crossings.is_empty() {
210 let mut last_flat_crossing_ts = None;
213 let mut running_qty = Decimal::ZERO;
214
215 for fill in fills {
216 let prev_qty = running_qty;
217 running_qty += Decimal::from(fill.direction()) * fill.qty;
218
219 if prev_qty != Decimal::ZERO && running_qty == Decimal::ZERO {
220 last_flat_crossing_ts = Some(fill.ts_event);
221 }
222 }
223
224 let lifecycle_boundary_ts =
225 last_flat_crossing_ts.unwrap_or(*zero_crossings.last().unwrap());
226
227 let current_lifecycle_fills: Vec<FillSnapshot> = fills
229 .iter()
230 .filter(|f| f.ts_event > lifecycle_boundary_ts)
231 .cloned()
232 .collect();
233
234 if current_lifecycle_fills.is_empty() {
235 return FillAdjustmentResult::NoAdjustment;
236 }
237
238 let (current_qty, current_value) = simulate_position(¤t_lifecycle_fills);
240
241 if check_position_match(
243 current_qty,
244 current_value,
245 venue_qty_signed,
246 venue_position.avg_px,
247 tolerance,
248 ) {
249 return FillAdjustmentResult::FilterToCurrentLifecycle {
251 last_zero_crossing_ts: lifecycle_boundary_ts,
252 current_lifecycle_fills,
253 };
254 }
255
256 if let Some(first_fill) = current_lifecycle_fills.first() {
258 let synthetic_fill = FillSnapshot::new(
259 first_fill.venue_order_id,
260 position_to_order_side(venue_position.side),
261 venue_position.qty,
262 venue_position.avg_px,
263 first_fill.ts_event.saturating_sub(1), );
265
266 return FillAdjustmentResult::ReplaceCurrentLifecycle {
267 synthetic_fill,
268 first_venue_order_id: first_fill.venue_order_id,
269 };
270 }
271
272 return FillAdjustmentResult::NoAdjustment;
273 }
274
275 let oldest_lifecycle_fills: Vec<FillSnapshot> =
278 if let Some(&first_zero_crossing_ts) = zero_crossings.first() {
279 fills
281 .iter()
282 .filter(|f| f.ts_event <= first_zero_crossing_ts)
283 .cloned()
284 .collect()
285 } else {
286 fills.to_vec()
288 };
289
290 if oldest_lifecycle_fills.is_empty() {
291 return FillAdjustmentResult::NoAdjustment;
292 }
293
294 let (oldest_qty, oldest_value) = simulate_position(&oldest_lifecycle_fills);
296
297 if zero_crossings.is_empty() {
299 if check_position_match(
301 oldest_qty,
302 oldest_value,
303 venue_qty_signed,
304 venue_position.avg_px,
305 tolerance,
306 ) {
307 return FillAdjustmentResult::NoAdjustment;
308 }
309
310 if let Some(first_fill) = oldest_lifecycle_fills.first() {
312 let oldest_avg_px = if oldest_qty == Decimal::ZERO {
315 None
316 } else {
317 Some(oldest_value / oldest_qty.abs())
318 };
319
320 let reconciliation_price = calculate_reconciliation_price(
321 oldest_qty,
322 oldest_avg_px,
323 venue_qty_signed,
324 Some(venue_position.avg_px),
325 );
326
327 if let Some(opening_px) = reconciliation_price {
328 let opening_qty = if oldest_qty == Decimal::ZERO {
330 venue_qty_signed
331 } else {
332 venue_qty_signed - oldest_qty
334 };
335
336 if opening_qty.abs() > Decimal::ZERO {
337 let synthetic_side = if opening_qty > Decimal::ZERO {
338 OrderSide::Buy
339 } else {
340 OrderSide::Sell
341 };
342
343 let synthetic_fill = FillSnapshot::new(
344 first_fill.venue_order_id,
345 synthetic_side,
346 opening_qty.abs(),
347 opening_px,
348 first_fill.ts_event.saturating_sub(1),
349 );
350
351 return FillAdjustmentResult::AddSyntheticOpening {
352 synthetic_fill,
353 existing_fills: oldest_lifecycle_fills,
354 };
355 }
356 }
357 }
358
359 return FillAdjustmentResult::NoAdjustment;
360 }
361
362 if oldest_qty == Decimal::ZERO {
364 return FillAdjustmentResult::NoAdjustment;
366 }
367
368 if !oldest_lifecycle_fills.is_empty()
370 && let Some(&first_zero_crossing_ts) = zero_crossings.first()
371 {
372 let current_lifecycle_fills: Vec<FillSnapshot> = fills
374 .iter()
375 .filter(|f| f.ts_event > first_zero_crossing_ts)
376 .cloned()
377 .collect();
378
379 if !current_lifecycle_fills.is_empty()
380 && let Some(first_current_fill) = current_lifecycle_fills.first()
381 {
382 let synthetic_fill = FillSnapshot::new(
383 first_current_fill.venue_order_id,
384 position_to_order_side(venue_position.side),
385 venue_position.qty,
386 venue_position.avg_px,
387 first_current_fill.ts_event.saturating_sub(1),
388 );
389
390 return FillAdjustmentResult::AddSyntheticOpening {
391 synthetic_fill,
392 existing_fills: oldest_lifecycle_fills,
393 };
394 }
395 }
396
397 FillAdjustmentResult::NoAdjustment
398}
399
400pub fn check_position_reconciliation(
411 report: &PositionStatusReport,
412 cached_signed_qty: Decimal,
413 size_precision: Option<u8>,
414) -> bool {
415 let venue_signed_qty = report.signed_decimal_qty;
416
417 if venue_signed_qty == Decimal::ZERO && cached_signed_qty == Decimal::ZERO {
418 return true;
419 }
420
421 if let Some(precision) = size_precision
422 && is_within_single_unit_tolerance(cached_signed_qty, venue_signed_qty, precision)
423 {
424 log::debug!(
425 "Position for {} within tolerance: cached={}, venue={}",
426 report.instrument_id,
427 cached_signed_qty,
428 venue_signed_qty
429 );
430 return true;
431 }
432
433 if cached_signed_qty == venue_signed_qty {
434 return true;
435 }
436
437 log::warn!(
438 "Position discrepancy for {}: cached={}, venue={}",
439 report.instrument_id,
440 cached_signed_qty,
441 venue_signed_qty
442 );
443
444 false
445}
446
447pub(super) fn cap_price_at_instrument_max(px: Decimal, instrument: &InstrumentAny) -> Decimal {
465 let Some(max) = instrument.max_price() else {
466 return px;
467 };
468 let cap = max.as_decimal().round_dp_with_strategy(
469 u32::from(instrument.price_precision()),
470 RoundingStrategy::ToZero,
471 );
472 px.min(cap)
473}
474
475pub(super) fn create_synthetic_order_report(
485 fill: &FillSnapshot,
486 account_id: AccountId,
487 instrument_id: InstrumentId,
488 instrument: &InstrumentAny,
489 venue_order_id: VenueOrderId,
490) -> anyhow::Result<OrderStatusReport> {
491 let order_qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
492
493 let mut report = OrderStatusReport::new(
494 account_id,
495 instrument_id,
496 None, venue_order_id,
498 fill.side,
499 OrderType::Market,
500 TimeInForce::Gtc,
501 OrderStatus::Filled,
502 order_qty,
503 order_qty, UnixNanos::from(fill.ts_event),
505 UnixNanos::from(fill.ts_event),
506 UnixNanos::from(fill.ts_event),
507 None, );
509 report.avg_px = Some(cap_price_at_instrument_max(fill.px, instrument));
510 Ok(report)
511}
512
513pub(super) fn create_synthetic_fill_report(
519 fill: &FillSnapshot,
520 account_id: AccountId,
521 instrument_id: InstrumentId,
522 instrument: &InstrumentAny,
523 venue_order_id: VenueOrderId,
524) -> anyhow::Result<FillReport> {
525 let trade_id = create_synthetic_trade_id(fill);
526 let qty = Quantity::from_decimal_dp(fill.qty, instrument.size_precision())?;
527 let px = Price::from_decimal_dp(
528 cap_price_at_instrument_max(fill.px, instrument),
529 instrument.price_precision(),
530 )?;
531
532 Ok(FillReport::new(
533 account_id,
534 instrument_id,
535 venue_order_id,
536 trade_id,
537 fill.side,
538 qty,
539 px,
540 Money::zero(instrument.quote_currency()),
541 LiquiditySide::NoLiquiditySide,
542 None, None, fill.ts_event.into(),
545 fill.ts_event.into(),
546 None, ))
548}
549
550fn position_report_to_snapshot(report: &PositionStatusReport) -> VenuePositionSnapshot {
552 VenuePositionSnapshot {
553 side: report.position_side,
554 qty: report.quantity.into(),
555 avg_px: report.avg_px_open.unwrap_or(Decimal::ZERO),
556 }
557}
558
559fn position_to_order_side(side: PositionSideSpecified) -> OrderSide {
565 match side {
566 PositionSideSpecified::Long => OrderSide::Buy,
567 PositionSideSpecified::Short => OrderSide::Sell,
568 PositionSideSpecified::Flat => {
569 unreachable!("flat venue position must be guarded by an earlier check")
570 }
571 }
572}
573
574fn extract_instrument_reports(
576 mass_status: &ExecutionMassStatus,
577 instrument_id: InstrumentId,
578) -> ReconciliationResult {
579 let mut orders = IndexMap::new();
580 let mut fills = IndexMap::new();
581
582 for (id, order) in mass_status.order_reports() {
583 if order.instrument_id == instrument_id {
584 orders.insert(id, order.clone());
585 }
586 }
587
588 for (id, fill_list) in mass_status.fill_reports() {
589 let filtered: Vec<_> = fill_list
590 .iter()
591 .filter(|f| f.instrument_id == instrument_id)
592 .cloned()
593 .collect();
594
595 if !filtered.is_empty() {
596 fills.insert(id, filtered);
597 }
598 }
599
600 ReconciliationResult { orders, fills }
601}
602
603struct ExtractedFills {
605 snapshots: Vec<FillSnapshot>,
606 orders: IndexMap<VenueOrderId, OrderStatusReport>,
607 fills: IndexMap<VenueOrderId, Vec<FillReport>>,
608}
609
610fn extract_fills_for_instrument(
612 mass_status: &ExecutionMassStatus,
613 instrument_id: InstrumentId,
614) -> ExtractedFills {
615 let mut snapshots = Vec::new();
616 let mut order_map = IndexMap::new();
617 let mut fill_map = IndexMap::new();
618
619 for (id, order) in mass_status.order_reports() {
621 if order.instrument_id == instrument_id {
622 order_map.insert(id, order.clone());
623 }
624 }
625
626 for (venue_order_id, fill_reports) in mass_status.fill_reports() {
628 for fill in fill_reports {
629 if fill.instrument_id == instrument_id {
630 let side = mass_status
631 .order_reports()
632 .get(&venue_order_id)
633 .map_or(fill.order_side, |o| o.order_side);
634
635 snapshots.push(FillSnapshot::new(
636 venue_order_id,
637 side,
638 fill.last_qty.into(),
639 fill.last_px.into(),
640 fill.ts_event.as_u64(),
641 ));
642
643 fill_map
644 .entry(venue_order_id)
645 .or_insert_with(Vec::new)
646 .push(fill.clone());
647 }
648 }
649 }
650
651 snapshots.sort_by_key(|f| f.ts_event);
653
654 ExtractedFills {
655 snapshots,
656 orders: order_map,
657 fills: fill_map,
658 }
659}
660
661#[must_use]
667pub(super) fn simulate_position(fills: &[FillSnapshot]) -> (Decimal, Decimal) {
668 let mut qty = Decimal::ZERO;
669 let mut value = Decimal::ZERO;
670
671 for fill in fills {
672 debug_assert!(
673 fill.qty > Decimal::ZERO,
674 "fill snapshot qty must be positive, received {}",
675 fill.qty,
676 );
677 let direction = Decimal::from(fill.direction());
678 let new_qty = qty + (direction * fill.qty);
679
680 if (qty >= Decimal::ZERO && direction > Decimal::ZERO)
682 || (qty <= Decimal::ZERO && direction < Decimal::ZERO)
683 {
684 value += fill.qty * fill.px;
686 qty = new_qty;
687 } else {
688 if qty.abs() >= fill.qty {
690 let close_ratio = fill.qty / qty.abs();
692 value *= Decimal::ONE - close_ratio;
693 qty = new_qty;
694 } else {
695 let remaining = fill.qty - qty.abs();
697 qty = direction * remaining;
698 value = remaining * fill.px;
699 }
700 }
701 }
702
703 debug_assert!(
704 value >= Decimal::ZERO,
705 "simulated position value must be non-negative, was {value}",
706 );
707 debug_assert!(
708 !(qty != Decimal::ZERO && value.is_sign_negative()),
709 "simulated avg price invariant: qty={qty}, value={value}",
710 );
711
712 (qty, value)
713}
714
715#[must_use]
724pub(super) fn detect_zero_crossings(fills: &[FillSnapshot]) -> Vec<u64> {
725 let mut running_qty = Decimal::ZERO;
726 let mut zero_crossings = Vec::new();
727
728 for fill in fills {
729 let prev_qty = running_qty;
730 running_qty += Decimal::from(fill.direction()) * fill.qty;
731
732 if prev_qty != Decimal::ZERO {
734 if running_qty == Decimal::ZERO {
735 zero_crossings.push(fill.ts_event);
737 } else if (prev_qty > Decimal::ZERO) != (running_qty > Decimal::ZERO) {
738 zero_crossings.push(fill.ts_event);
740 }
741 }
742 }
743
744 zero_crossings
745}
746
747#[must_use]
753pub(super) fn check_position_match(
754 simulated_qty: Decimal,
755 simulated_value: Decimal,
756 venue_qty: Decimal,
757 venue_avg_px: Decimal,
758 tolerance: Decimal,
759) -> bool {
760 if simulated_qty != venue_qty {
761 return false;
762 }
763
764 if simulated_qty == Decimal::ZERO {
765 return true; }
767
768 let abs_qty = simulated_qty.abs();
770 if abs_qty == Decimal::ZERO {
771 return false;
772 }
773
774 let simulated_avg_px = simulated_value / abs_qty;
775
776 if venue_avg_px == Decimal::ZERO {
778 return false;
779 }
780
781 let relative_diff = (simulated_avg_px - venue_avg_px).abs() / venue_avg_px;
782
783 relative_diff <= tolerance
784}
785
786pub fn calculate_reconciliation_price(
804 current_position_qty: Decimal,
805 current_position_avg_px: Option<Decimal>,
806 target_position_qty: Decimal,
807 target_position_avg_px: Option<Decimal>,
808) -> Option<Decimal> {
809 let qty_diff = target_position_qty - current_position_qty;
810
811 if qty_diff == Decimal::ZERO {
812 return None; }
814
815 if target_position_qty == Decimal::ZERO {
818 return current_position_avg_px;
819 }
820
821 let target_avg_px = target_position_avg_px?;
823 if target_avg_px == Decimal::ZERO {
824 return None;
825 }
826
827 if current_position_qty == Decimal::ZERO || current_position_avg_px.is_none() {
829 return Some(target_avg_px);
830 }
831
832 let current_avg_px = current_position_avg_px?;
833
834 let is_flip = (current_position_qty > Decimal::ZERO) != (target_position_qty > Decimal::ZERO)
837 && target_position_qty != Decimal::ZERO;
838
839 if is_flip {
840 return Some(target_avg_px);
841 }
842
843 let target_value = target_position_qty * target_avg_px;
846 let current_value = current_position_qty * current_avg_px;
847 let diff_value = target_value - current_value;
848
849 let reconciliation_px = diff_value / qty_diff;
851
852 if reconciliation_px > Decimal::ZERO {
854 return Some(reconciliation_px);
855 }
856
857 None
858}
859
860#[must_use]
865pub(super) fn is_within_single_unit_tolerance(
866 value1: Decimal,
867 value2: Decimal,
868 precision: u8,
869) -> bool {
870 if precision == 0 {
871 return value1 == value2;
872 }
873
874 let tolerance = Decimal::new(1, u32::from(precision));
875 let difference = (value1 - value2).abs();
876 difference <= tolerance
877}