Skip to main content

nautilus_execution/reconciliation/
positions.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Position-state reconciliation.
17//!
18//! Position simulation, partial-window fill reconstruction, mass-status processing,
19//! and final position-match checks. The core invariant maintained here is that the
20//! reconstructed position matches the venue's reported position within tolerance
21//! (default 0.01%) after reconciliation is applied.
22
23use 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); // 0.0001
40
41/// Process fill reports from a mass status for position reconciliation.
42///
43/// This is the main entry point for position reconciliation. It:
44/// 1. Extracts fills and position for the given instrument
45/// 2. Detects position discrepancies
46/// 3. Returns adjusted order/fill reports ready for processing
47///
48/// # Errors
49///
50/// Returns an error if synthetic report creation fails.
51pub 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
59/// Process fill reports without generating synthetic order or fill reports.
60///
61/// Non-generating adjustments, such as filtering completed position lifecycles, are still applied.
62///
63/// # Errors
64///
65/// Returns an error if report processing fails.
66pub 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    // Get position report for this instrument
85    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            // No position report - return orders/fills unchanged
90            return Ok(extract_instrument_reports(mass_status, instrument_id));
91        }
92    };
93
94    // Extract and convert fills to snapshots
95    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    // Run adjustment logic
108    let result = adjust_fills_for_partial_window(&fill_snapshots, &venue_position, tol);
109
110    // Apply adjustments
111    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            // Replace with only synthetic
158            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            // Filter fills to current lifecycle
169            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            // Keep only orders that have fills or are still working
175            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/// Adjust fills for partial reconciliation window to handle incomplete position lifecycles.
201///
202/// This function analyzes fills and determines if adjustments are needed when the reconciliation
203/// window doesn't capture the complete position history (missing opening fills).
204///
205/// # Returns
206///
207/// Returns `FillAdjustmentResult` indicating what adjustments (if any) are needed.
208#[must_use]
209pub(super) fn adjust_fills_for_partial_window(
210    fills: &[FillSnapshot],
211    venue_position: &VenuePositionSnapshot,
212    tolerance: Decimal,
213) -> FillAdjustmentResult {
214    // If no fills, nothing to adjust
215    if fills.is_empty() {
216        return FillAdjustmentResult::NoAdjustment;
217    }
218
219    // If venue position is FLAT, return unchanged
220    if venue_position.qty == Decimal::ZERO {
221        return FillAdjustmentResult::NoAdjustment;
222    }
223
224    // Detect zero-crossings
225    let zero_crossings = detect_zero_crossings(fills);
226
227    // Convert venue position to signed quantity
228    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    // Case 1: Has zero-crossings - focus on current lifecycle after last zero-crossing
235    if !zero_crossings.is_empty() {
236        // Find the last zero-crossing that lands on FLAT (qty==0)
237        // This separates lifecycles; flips within a lifecycle don't count
238        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        // Get fills from current lifecycle (after lifecycle boundary)
254        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        // Simulate current lifecycle
265        let (current_qty, current_value) = simulate_position(&current_lifecycle_fills);
266
267        // Check if current lifecycle matches venue
268        if check_position_match(
269            current_qty,
270            current_value,
271            venue_qty_signed,
272            venue_position.avg_px,
273            tolerance,
274        ) {
275            // Current lifecycle matches - filter out old lifecycles
276            return FillAdjustmentResult::FilterToCurrentLifecycle {
277                last_zero_crossing_ts: lifecycle_boundary_ts,
278                current_lifecycle_fills,
279            };
280        }
281
282        // Current lifecycle doesn't match - replace with synthetic fill
283        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), // Timestamp before first fill
290            );
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    // Case 2: Single lifecycle or one zero-crossing
302    // Determine which fills to analyze
303    let oldest_lifecycle_fills: Vec<FillSnapshot> =
304        if let Some(&first_zero_crossing_ts) = zero_crossings.first() {
305            // Get fills before first zero-crossing
306            fills
307                .iter()
308                .filter(|f| f.ts_event <= first_zero_crossing_ts)
309                .cloned()
310                .collect()
311        } else {
312            // No zero-crossings - all fills are in single lifecycle
313            fills.to_vec()
314        };
315
316    if oldest_lifecycle_fills.is_empty() {
317        return FillAdjustmentResult::NoAdjustment;
318    }
319
320    // Simulate oldest lifecycle
321    let (oldest_qty, oldest_value) = simulate_position(&oldest_lifecycle_fills);
322
323    // If single lifecycle (no zero-crossings)
324    if zero_crossings.is_empty() {
325        // Check if simulated position matches venue
326        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        // Doesn't match - need to add synthetic opening fill
337        if let Some(first_fill) = oldest_lifecycle_fills.first() {
338            // Calculate what opening fill is needed
339            // Use simulated position as current, venue position as target
340            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                // Calculate opening quantity needed
355                let opening_qty = if oldest_qty == Decimal::ZERO {
356                    venue_qty_signed
357                } else {
358                    // Work backwards: venue = opening + current fills
359                    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    // Has one zero-crossing - check if oldest lifecycle closes at zero
389    if oldest_qty == Decimal::ZERO {
390        // Lifecycle closes correctly - no adjustment needed
391        return FillAdjustmentResult::NoAdjustment;
392    }
393
394    // Oldest lifecycle doesn't close at zero - add synthetic opening fill
395    if !oldest_lifecycle_fills.is_empty()
396        && let Some(&first_zero_crossing_ts) = zero_crossings.first()
397    {
398        // Need to add opening fill that makes position close at zero-crossing
399        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
426/// Check if a cached position matches the venue position report within tolerance.
427///
428/// Compares the cached signed quantity against the venue-reported signed quantity,
429/// allowing a single-unit tolerance at the instrument's size precision when provided.
430///
431/// # Returns
432///
433/// Returns `true` when positions match (within tolerance), `false` otherwise. Logs
434/// a debug message when values differ but fall within tolerance, and a warning
435/// when they fail to reconcile.
436pub 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
473/// Caps a price at the instrument's maximum price.
474///
475/// Reconciliation derives synthetic prices by dividing a notional by a
476/// quantity; when that quantity is a dust rounding residual the result blows up
477/// far above the instrument's tradeable range. Capping at `max_price` keeps
478/// synthetic reports, and the positions derived from them, within range.
479/// Because the blown-up price always pairs with a dust quantity (the
480/// denominator) the PnL impact is negligible. Only the upper bound is enforced:
481/// the blow-up is always on the high side, so the lower bound is left untouched
482/// and legitimate negative-price and zero-cost fills pass through unchanged. A
483/// missing `max_price` leaves the price unmodified.
484///
485/// The cap is `max_price` floored to the instrument's price precision, so that
486/// rebuilding a `Price` at that precision (which rounds) cannot lift the result
487/// back above `max_price`. For example a 2 dp instrument with `max_price = 0.999`
488/// caps at `0.99`, since `0.999` (or any value in `(0.99, 1.00)`) rebuilt at 2 dp
489/// would round to `1.00`.
490pub(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
501/// Create a synthetic `OrderStatusReport` from a `FillSnapshot`.
502///
503/// Populates `avg_px` from the fill's price so downstream reconciliation paths
504/// (e.g. [`crate::reconciliation::orders::create_inferred_fill`]) can resolve a
505/// fill price without falling back to the "no `avg_px`, report price, or order price" warning.
506///
507/// # Errors
508///
509/// Returns an error if the fill quantity cannot be converted to f64.
510pub(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, // client_order_id
523        venue_order_id,
524        fill.side.into(),
525        OrderType::Market,
526        TimeInForce::Gtc,
527        OrderStatus::Filled,
528        order_qty,
529        order_qty, // filled_qty = order_qty (fully filled)
530        UnixNanos::from(fill.ts_event),
531        UnixNanos::from(fill.ts_event),
532        UnixNanos::from(fill.ts_event),
533        None, // report_id
534    );
535    report.avg_px = Some(cap_price_at_instrument_max(fill.px, instrument));
536    Ok(report)
537}
538
539/// Create a synthetic `FillReport` from a `FillSnapshot`.
540///
541/// # Errors
542///
543/// Returns an error if the fill quantity or price cannot be converted.
544pub(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, // client_order_id
569        None, // venue_position_id
570        fill.ts_event.into(),
571        fill.ts_event.into(),
572        None, // report_id
573    ))
574}
575
576/// Convert a position status report to a venue position snapshot.
577fn 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
585/// Map a position side to the order side that would open it.
586///
587/// Callers must guard `Flat` upstream; reaching it here means a flat venue
588/// position slipped past the qty==0 early returns in
589/// [`adjust_fills_for_partial_window`].
590fn 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
600/// Extract orders and fills for a specific instrument from mass status.
601fn 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
629/// Extracted fills and reports for an instrument.
630struct ExtractedFills {
631    snapshots: Vec<FillSnapshot>,
632    orders: IndexMap<VenueOrderId, OrderStatusReport>,
633    fills: IndexMap<VenueOrderId, Vec<FillReport>>,
634}
635
636/// Extract fills for an instrument and convert to snapshots.
637fn 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    // Seed order_map
646    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    // Extract fills
653    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    // Sort chronologically
679    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/// Simulate position from chronologically ordered fills using netting logic.
689///
690/// # Returns
691///
692/// Returns a tuple of (quantity, value) after applying all fills.
693#[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        // Check if we're accumulating or crossing zero (flip/close)
708        if (qty >= Decimal::ZERO && direction > Decimal::ZERO)
709            || (qty <= Decimal::ZERO && direction < Decimal::ZERO)
710        {
711            // Accumulating in same direction
712            value += fill.qty * fill.px;
713            qty = new_qty;
714        } else {
715            // Closing or flipping position
716            if qty.abs() >= fill.qty {
717                // Partial close - maintain average price by reducing value proportionally
718                let close_ratio = fill.qty / qty.abs();
719                value *= Decimal::ONE - close_ratio;
720                qty = new_qty;
721            } else {
722                // Close and flip - reset value to opening position
723                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/// Detect zero-crossing timestamps in a sequence of fills.
743///
744/// A zero-crossing occurs when position quantity crosses through zero (FLAT).
745/// This includes both landing exactly on zero and flipping from long to short or vice versa.
746///
747/// # Returns
748///
749/// Returns a list of timestamps where position crosses through zero.
750#[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        // Detect when position crosses zero
760        if prev_qty != Decimal::ZERO {
761            if running_qty == Decimal::ZERO {
762                // Landed exactly on zero
763                zero_crossings.push(fill.ts_event);
764            } else if (prev_qty > Decimal::ZERO) != (running_qty > Decimal::ZERO) {
765                // Sign changed - crossed through zero (flip)
766                zero_crossings.push(fill.ts_event);
767            }
768        }
769    }
770
771    zero_crossings
772}
773
774/// Check if simulated position matches venue position within tolerance.
775///
776/// # Returns
777///
778/// Returns true if quantities and average prices match within tolerance.
779#[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; // Both FLAT
793    }
794
795    // Guard against division by zero
796    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 is zero, we cannot calculate relative difference
804    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
813/// Calculate the price needed for a reconciliation order to achieve target position.
814///
815/// This is a pure function that calculates what price a fill would need to have
816/// to move from the current position state to the target position state with the
817/// correct average price, accounting for the netting simulation logic.
818///
819/// # Returns
820///
821/// Returns `Some(Decimal)` if a valid reconciliation price can be calculated, `None` otherwise.
822///
823/// # Notes
824///
825/// The function handles four scenarios:
826/// 1. Position to flat: `reconciliation_px` = `current_avg_px` (close at current average)
827/// 2. Flat to position: `reconciliation_px` = `target_avg_px`
828/// 3. Position flip (sign change): `reconciliation_px` = `target_avg_px` (due to value reset in simulation)
829/// 4. Accumulation/reduction: weighted average formula
830pub 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; // No reconciliation needed
840    }
841
842    // Special case: closing to flat (target_position_qty == 0)
843    // When flattening, the reconciliation price equals the current position's average price
844    if target_position_qty == Decimal::ZERO {
845        return current_position_avg_px;
846    }
847
848    // If target average price is not provided or zero, we cannot calculate
849    let target_avg_px = target_position_avg_px?;
850    if target_avg_px == Decimal::ZERO {
851        return None;
852    }
853
854    // If current position is flat, the reconciliation price equals target avg price
855    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    // Check if this is a flip scenario (sign change)
862    // In simulation, flips reset value to remaining * px, so reconciliation_px = target_avg_px
863    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    // For accumulation or reduction (same side), use weighted average formula
871    // Formula: (target_qty * target_avg_px) = (current_qty * current_avg_px) + (qty_diff * reconciliation_px)
872    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    // qty_diff is guaranteed non-zero by the early return above
877    let reconciliation_px = diff_value / qty_diff;
878
879    // Ensure price is positive
880    if reconciliation_px > Decimal::ZERO {
881        return Some(reconciliation_px);
882    }
883
884    None
885}
886
887/// Checks if two decimal values are within a single unit of tolerance for the given precision.
888///
889/// For integer precision (0), requires exact match.
890/// For fractional precision, allows difference of 1 unit at that precision.
891#[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}