diff --git a/crates/common/src/cache/mod.rs b/crates/common/src/cache/mod.rs index f7216c51e8d4..3532235fc8f8 100644 --- a/crates/common/src/cache/mod.rs +++ b/crates/common/src/cache/mod.rs @@ -2347,7 +2347,8 @@ impl Cache { .collect(); if let Some(db) = &self.database { - self.index.order_position = db.load_index_order_position()?; + let order_position = db.load_index_order_position()?; + self.index.order_position = self.sanitize_order_position_index(order_position); self.index.order_client = db.load_index_order_client()?; } @@ -2444,7 +2445,8 @@ impl Cache { }; if let Some(db) = &self.database { - self.index.order_position = db.load_index_order_position()?; + let order_position = db.load_index_order_position()?; + self.index.order_position = self.sanitize_order_position_index(order_position); self.index.order_client = db.load_index_order_client()?; } @@ -2454,6 +2456,23 @@ impl Cache { Ok(()) } + fn sanitize_order_position_index( + &self, + mut order_position: AHashMap, + ) -> AHashMap { + let original_len = order_position.len(); + order_position.retain(|client_order_id, _| self.orders.contains_key(client_order_id)); + let removed = original_len - order_position.len(); + + if removed > 0 { + log::warn!( + "Filtered {removed} stale order-position index entries without backing orders during cache load" + ); + } + + order_position + } + /// Clears and reloads the position cache from the database. /// /// # Errors @@ -2644,11 +2663,13 @@ impl Cache { .insert(*position_id, position.strategy_id); // 3: Build index.position_orders -> {PositionId, {ClientOrderId}} - self.index - .position_orders - .entry(*position_id) - .or_default() - .extend(position.client_order_ids()); + let position_orders = self.index.position_orders.entry(*position_id).or_default(); + position_orders.extend( + position + .client_order_ids() + .into_iter() + .filter(|client_order_id| self.orders.contains_key(client_order_id)), + ); // 4: Build index.instrument_positions -> {InstrumentId, {PositionId}} self.index @@ -2656,6 +2677,10 @@ impl Cache { .entry(instrument_id) .or_default() .insert(*position_id); + self.index + .instrument_orders + .entry(instrument_id) + .or_default(); // 5: Build index.strategy_positions -> {StrategyId, {PositionId}} self.index @@ -2663,6 +2688,7 @@ impl Cache { .entry(strategy_id) .or_default() .insert(*position_id); + self.index.strategy_orders.entry(strategy_id).or_default(); // 6: Build index.account_positions -> {AccountId, {PositionId}} self.index @@ -4632,18 +4658,30 @@ impl Cache { database.index_order_position(*client_order_id, *position_id)?; } - // Index: PositionId -> StrategyId - self.index - .position_strategy - .insert(*position_id, *strategy_id); - - // Index: PositionId -> set[ClientOrderId] + self.index_position(position_id, venue, strategy_id); self.index .position_orders .entry(*position_id) .or_default() .insert(*client_order_id); + Ok(()) + } + + fn index_position( + &mut self, + position_id: &PositionId, + venue: &Venue, + strategy_id: &StrategyId, + ) { + // Index: PositionId -> StrategyId + self.index + .position_strategy + .insert(*position_id, *strategy_id); + + // Every position has a reverse-order bucket, including orderless positions. + self.index.position_orders.entry(*position_id).or_default(); + // Index: StrategyId -> set[PositionId] self.index .strategy_positions @@ -4657,8 +4695,6 @@ impl Cache { .entry(*venue) .or_default() .insert(*position_id); - - Ok(()) } // Propagates parent OTO `position_id` to contingent children that are missing one. @@ -4722,21 +4758,56 @@ impl Cache { /// /// Returns an error if persisting the position to the backing database fails. pub fn add_position(&mut self, position: &Position, oms_type: OmsType) -> anyhow::Result<()> { + self.add_position_inner(position, oms_type, true) + } + + /// Adds a position whose opening fill intentionally has no backing order. + /// + /// # Errors + /// + /// Returns an error if persisting the position to the backing database fails. + pub fn add_position_without_order( + &mut self, + position: &Position, + oms_type: OmsType, + ) -> anyhow::Result<()> { + self.add_position_inner(position, oms_type, false) + } + + fn add_position_inner( + &mut self, + position: &Position, + oms_type: OmsType, + index_order: bool, + ) -> anyhow::Result<()> { self.positions .insert(position.id, SharedCell::new(position.clone())); self.index.position_oms.insert(position.id, oms_type); self.index.positions.insert(position.id); self.index.positions_open.insert(position.id); self.index.positions_closed.remove(&position.id); // Cleanup for NETTING reopen + self.index.strategies.insert(position.strategy_id); + self.index + .strategy_orders + .entry(position.strategy_id) + .or_default(); log::debug!("Adding {position}"); - self.add_position_id( - &position.id, - &position.instrument_id.venue, - &position.opening_order_id, - &position.strategy_id, - )?; + if index_order { + self.add_position_id( + &position.id, + &position.instrument_id.venue, + &position.opening_order_id, + &position.strategy_id, + )?; + } else { + self.index_position( + &position.id, + &position.instrument_id.venue, + &position.strategy_id, + ); + } let venue = position.instrument_id.venue; let venue_positions = self.index.venue_positions.entry(venue).or_default(); @@ -4750,6 +4821,10 @@ impl Cache { .entry(instrument_id) .or_default(); instrument_positions.insert(position.id); + self.index + .instrument_orders + .entry(instrument_id) + .or_default(); // Index: AccountId -> AHashSet self.index diff --git a/crates/common/src/cache/tests.rs b/crates/common/src/cache/tests.rs index 47923f14b4ff..1f735890d573 100644 --- a/crates/common/src/cache/tests.rs +++ b/crates/common/src/cache/tests.rs @@ -476,6 +476,79 @@ fn test_build_index_restores_bidirectional_venue_order_id_lookup( ); } +#[rstest] +fn test_build_index_preserves_orderless_position_strategy_bucket( + mut cache: Cache, + audusd_sim: CurrencyPair, +) { + let instrument = InstrumentAny::CurrencyPair(audusd_sim); + let order = OrderTestBuilder::new(OrderType::Market) + .instrument_id(instrument.id()) + .side(OrderSide::Buy) + .quantity(Quantity::from(100_000)) + .build(); + let client_order_id = order.client_order_id(); + let strategy_id = order.strategy_id(); + let position_id = PositionId::new("P-ORDERLESS-BUILD-INDEX"); + let fill = TestOrderEventStubs::filled( + &order, + &instrument, + Some(TradeId::new("T-ORDERLESS-BUILD-INDEX")), + Some(position_id), + Some(Price::from("1.00001")), + None, + None, + None, + None, + None, + ); + let position = Position::new(&instrument, fill.into()); + + cache + .add_position_without_order(&position, OmsType::Netting) + .unwrap(); + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); + assert!( + cache + .index + .strategy_orders + .get(&strategy_id) + .is_some_and(|orders| orders.is_empty()) + ); + assert!( + cache + .index + .position_orders + .get(&position_id) + .is_some_and(|orders| orders.is_empty()) + ); + assert!(cache.orders_for_position(&position_id).is_empty()); + assert!(cache.check_integrity()); + + cache.clear_index(); + cache.build_index(); + + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); + assert!( + cache + .index + .strategy_orders + .get(&strategy_id) + .is_some_and(|orders| orders.is_empty()) + ); + assert!( + cache + .index + .position_orders + .get(&position_id) + .is_some_and(|orders| orders.is_empty()) + ); + assert!(cache.orders_for_position(&position_id).is_empty()); + assert!(cache.check_integrity()); +} + #[rstest] fn test_oms_type_returns_actual_position_oms(mut cache: Cache) { let position = snapshot_test_position(); @@ -550,8 +623,7 @@ fn test_cache_positions_skips_malformed_position_oms() { Bytes::from_static(b"invalid"), )]), positions: AHashMap::from([(position_id, position)]), - fail_add: false, - fail_update_order: false, + ..Default::default() }; let mut cache = Cache::default(); cache.set_database(Box::new(database)); @@ -935,6 +1007,78 @@ fn test_cache_orders_when_no_database(mut cache: Cache) { assert!(futures::executor::block_on(cache.cache_orders()).is_ok()); } +#[rstest] +fn test_cache_all_filters_legacy_order_position_without_backing_order(audusd_sim: CurrencyPair) { + let valid_order_id = ClientOrderId::from("O-VALID"); + let stale_order_id = ClientOrderId::from("SPREAD-LEG-STALE"); + let valid_position_id = PositionId::from("P-VALID"); + let stale_position_id = PositionId::from("P-STALE"); + let mut order = OrderTestBuilder::new(OrderType::Market) + .instrument_id(audusd_sim.id) + .client_order_id(valid_order_id) + .side(OrderSide::Buy) + .quantity(Quantity::from(100_000)) + .build(); + order.set_position_id(Some(valid_position_id)); + let database = SnapshotBlobTestDatabase { + orders: AHashMap::from([(valid_order_id, order)]), + order_positions: AHashMap::from([ + (valid_order_id, valid_position_id), + (stale_order_id, stale_position_id), + ]), + ..Default::default() + }; + let mut cache = Cache::new(None, Some(Box::new(database))); + + futures::executor::block_on(cache.cache_all()).expect("cache all"); + + assert_eq!(cache.position_id(&valid_order_id), Some(&valid_position_id)); + assert_eq!(cache.position_id(&stale_order_id), None); + cache.build_index(); + assert_eq!(cache.position_id(&stale_order_id), None); + assert!(cache.check_integrity()); + + futures::executor::block_on(cache.cache_all()).expect("cache all again"); + assert_eq!(cache.position_id(&valid_order_id), Some(&valid_position_id)); + assert_eq!(cache.position_id(&stale_order_id), None); +} + +#[rstest] +fn test_cache_orders_filters_legacy_order_position_without_backing_order(audusd_sim: CurrencyPair) { + let valid_order_id = ClientOrderId::from("O-VALID"); + let stale_order_id = ClientOrderId::from("SPREAD-LEG-STALE"); + let valid_position_id = PositionId::from("P-VALID"); + let stale_position_id = PositionId::from("P-STALE"); + let mut order = OrderTestBuilder::new(OrderType::Market) + .instrument_id(audusd_sim.id) + .client_order_id(valid_order_id) + .side(OrderSide::Buy) + .quantity(Quantity::from(100_000)) + .build(); + order.set_position_id(Some(valid_position_id)); + let database = SnapshotBlobTestDatabase { + orders: AHashMap::from([(valid_order_id, order)]), + order_positions: AHashMap::from([ + (valid_order_id, valid_position_id), + (stale_order_id, stale_position_id), + ]), + ..Default::default() + }; + let mut cache = Cache::new(None, Some(Box::new(database))); + + futures::executor::block_on(cache.cache_orders()).expect("cache orders"); + + assert_eq!(cache.position_id(&valid_order_id), Some(&valid_position_id)); + assert_eq!(cache.position_id(&stale_order_id), None); + cache.build_index(); + assert_eq!(cache.position_id(&stale_order_id), None); + assert!(cache.check_integrity()); + + futures::executor::block_on(cache.cache_orders()).expect("cache orders again"); + assert_eq!(cache.position_id(&valid_order_id), Some(&valid_position_id)); + assert_eq!(cache.position_id(&stale_order_id), None); +} + #[rstest] fn test_assign_position_ids_to_contingencies_propagates_parent_to_children( mut cache: Cache, @@ -6991,7 +7135,9 @@ fn snapshot_test_position() -> Position { #[derive(Default)] struct SnapshotBlobTestDatabase { general: AHashMap, + orders: AHashMap, positions: AHashMap, + order_positions: AHashMap, fail_add: bool, fail_update_order: bool, } @@ -7002,9 +7148,7 @@ impl SnapshotBlobTestDatabase { general.insert(key, value); Self { general, - positions: AHashMap::new(), - fail_add: false, - fail_update_order: false, + ..Default::default() } } @@ -7018,17 +7162,14 @@ impl SnapshotBlobTestDatabase { Self { general, positions, - fail_add: false, - fail_update_order: false, + ..Default::default() } } fn fail_add() -> Self { Self { - general: AHashMap::new(), - positions: AHashMap::new(), fail_add: true, - fail_update_order: false, + ..Default::default() } } @@ -7051,7 +7192,11 @@ impl CacheDatabaseAdapter for SnapshotBlobTestDatabase { } async fn load_all(&self) -> anyhow::Result { - Ok(CacheMap::default()) + Ok(CacheMap { + orders: self.orders.clone(), + positions: self.positions.clone(), + ..Default::default() + }) } fn load(&self) -> anyhow::Result> { @@ -7075,7 +7220,7 @@ impl CacheDatabaseAdapter for SnapshotBlobTestDatabase { } async fn load_orders(&self) -> anyhow::Result> { - Ok(AHashMap::new()) + Ok(self.orders.clone()) } async fn load_positions(&self) -> anyhow::Result> { @@ -7083,7 +7228,7 @@ impl CacheDatabaseAdapter for SnapshotBlobTestDatabase { } fn load_index_order_position(&self) -> anyhow::Result> { - Ok(AHashMap::new()) + Ok(self.order_positions.clone()) } fn load_index_order_client(&self) -> anyhow::Result> { diff --git a/crates/event_store/src/replay.rs b/crates/event_store/src/replay.rs index 2290c71a6488..f37156ee6721 100644 --- a/crates/event_store/src/replay.rs +++ b/crates/event_store/src/replay.rs @@ -616,6 +616,122 @@ impl CacheReplayError { } } +#[derive(Default)] +struct CacheReplayContext { + allow_deferred_orderless_flips: bool, + pending_orderless_flips: Vec, +} + +struct PendingOrderlessFlip { + source_seq: u64, + source_position_id: PositionId, + oms_type: OmsType, + opening_fill: OrderFilled, +} + +impl CacheReplayContext { + fn for_snapshot_tail() -> Self { + Self { + allow_deferred_orderless_flips: true, + pending_orderless_flips: Vec::new(), + } + } + + fn contains_flip_source(&self, fill: &OrderFilled) -> bool { + self.pending_orderless_flips.iter().any(|pending| { + pending.opening_fill.client_order_id == fill.client_order_id + && pending.opening_fill.trade_id == fill.trade_id + && pending.opening_fill.causation_id == Some(fill.event_id) + }) + } + + fn push_orderless_flip( + &mut self, + source_seq: u64, + source_position_id: PositionId, + oms_type: OmsType, + opening_fill: OrderFilled, + ) { + self.pending_orderless_flips.push(PendingOrderlessFlip { + source_seq, + source_position_id, + oms_type, + opening_fill, + }); + } + + fn take_opening_fill( + &mut self, + entry: &EventStoreEntry, + opened: &PositionOpened, + ) -> Result, CacheReplayError> { + let matching: Vec = self + .pending_orderless_flips + .iter() + .enumerate() + .filter_map(|(index, pending)| { + let fill = &pending.opening_fill; + (fill.trader_id == opened.trader_id + && fill.strategy_id == opened.strategy_id + && fill.instrument_id == opened.instrument_id + && fill.account_id == opened.account_id + && fill.client_order_id == opened.opening_order_id + && fill.order_side == opened.entry + && fill.last_qty == opened.last_qty + && fill.last_px == opened.last_px + && fill.currency == opened.currency) + .then_some(index) + }) + .collect(); + + if matching.len() > 1 { + return Err(apply_error( + entry, + format!( + "ambiguous orderless flip recovery for position {}: {} pending fills match", + opened.position_id, + matching.len(), + ), + )); + } + + let Some(index) = matching.first().copied() else { + return Ok(None); + }; + let pending = self.pending_orderless_flips.remove(index); + let mut fill = pending.opening_fill; + let oms_type = if pending.source_position_id == opened.position_id { + pending.oms_type + } else { + // A virtual HEDGING flip is the only live orderless path which opens a + // replacement ID. This recovers the OMS metadata even during a full + // event-store replay where no cache snapshot supplied it. + OmsType::Hedging + }; + fill.position_id = Some(opened.position_id); + // The live opening fragment is not stored separately. The PositionOpened event ID is + // stable and identifies the recovered opening transition, while causation still points + // to the original unsplit venue fill. + fill.event_id = opened.event_id; + Ok(Some((fill, oms_type))) + } + + fn ensure_complete(&self) -> Result<(), CacheReplayError> { + let Some(pending) = self.pending_orderless_flips.first() else { + return Ok(()); + }; + + Err(CacheReplayError::Apply { + seq: pending.source_seq, + payload_type: PAYLOAD_TYPE_ORDER_FILLED.to_string(), + message: format!( + "orderless flip fill {} has no matching PositionOpened event", + pending.opening_fill.trade_id, + ), + }) + } +} + /// Replays the cache snapshot tail after the caller restores the cache-owned snapshot blob. /// /// The restore hook runs before the tail iterator is consumed. When `anchor` is `Some`, @@ -647,6 +763,7 @@ where let mut applied_entries = 0; let mut ignored_entries = 0; + let mut context = CacheReplayContext::for_snapshot_tail(); for entry in scan { let entry = entry?; @@ -658,13 +775,15 @@ where }); } - if apply_cache_replay_entry(cache, &entry)? { + if apply_cache_replay_entry_with_context(cache, &entry, &mut context)? { applied_entries += 1; } else { ignored_entries += 1; } } + context.ensure_complete()?; + Ok(CacheReplayReport { plan, applied_entries, @@ -1123,6 +1242,17 @@ pub fn restore_cache_snapshot_blob( pub fn apply_cache_replay_entry( cache: &mut Cache, entry: &EventStoreEntry, +) -> Result { + let mut context = CacheReplayContext::default(); + let applied = apply_cache_replay_entry_with_context(cache, entry, &mut context)?; + context.ensure_complete()?; + Ok(applied) +} + +fn apply_cache_replay_entry_with_context( + cache: &mut Cache, + entry: &EventStoreEntry, + context: &mut CacheReplayContext, ) -> Result { if apply_complete_cache_payload_entry(cache, entry)? { return Ok(true); @@ -1180,29 +1310,18 @@ pub fn apply_cache_replay_entry( PAYLOAD_TYPE_ORDER_UPDATED => { apply_order_event(cache, entry, OrderEventAny::Updated)?; } - PAYLOAD_TYPE_ORDER_FILLED => { - let fill = decode_payload::(entry)?; - // The fill side panics deep inside Position/Order application; the hash - // proves the bytes match what was written, not that the producer wrote a - // semantically valid fill, so guard before the model invariants fire. - if matches!(fill.order_side, OrderSide::NoOrderSide) { - return Err(apply_error( - entry, - "OrderFilled.order_side must be Buy or Sell, was NoOrderSide", - )); - } - let event = OrderEventAny::Filled(fill.clone()); - apply_result(entry, cache.update_order(&event))?; - if !apply_fill_to_position(cache, entry, &fill)? { - return Ok(false); - } - } + PAYLOAD_TYPE_ORDER_FILLED => return apply_order_filled(cache, entry, context), PAYLOAD_TYPE_ORDER_FILL_VOIDED => { let fill_voided = decode_payload::(entry)?; apply_fill_void_to_order_and_positions(cache, entry, &fill_voided)?; } PAYLOAD_TYPE_POSITION_OPENED => { let opened = decode_payload::(entry)?; + if let Some(applied) = + apply_pending_orderless_flip_opened(cache, entry, &opened, context)? + { + return Ok(applied); + } return apply_position_opened(cache, entry, &opened); } PAYLOAD_TYPE_POSITION_CHANGED => { @@ -1295,6 +1414,113 @@ where Ok(wrap(decode_payload(entry)?)) } +fn apply_order_filled( + cache: &mut Cache, + entry: &EventStoreEntry, + context: &mut CacheReplayContext, +) -> Result { + let fill = decode_payload::(entry)?; + // The fill side panics deep inside Position/Order application; the hash + // proves the bytes match what was written, not that the producer wrote a + // semantically valid fill, so guard before the model invariants fire. + if matches!(fill.order_side, OrderSide::NoOrderSide) { + return Err(apply_error( + entry, + "OrderFilled.order_side must be Buy or Sell, was NoOrderSide", + )); + } + let event = OrderEventAny::Filled(fill.clone()); + let orderless_leg_fill = is_orderless_leg_fill(cache, &fill); + if !orderless_leg_fill { + apply_result(entry, cache.update_order(&event))?; + } + + let flip_applied = + orderless_leg_fill && apply_orderless_flip_fill(cache, entry, &fill, context)?; + + if flip_applied { + return Ok(true); + } + + apply_fill_to_position(cache, entry, &fill, orderless_leg_fill) +} + +fn apply_orderless_flip_fill( + cache: &mut Cache, + entry: &EventStoreEntry, + fill: &OrderFilled, + context: &mut CacheReplayContext, +) -> Result { + if context.contains_flip_source(fill) { + return Ok(true); + } + + let Some(position_id) = fill.position_id else { + return Ok(false); + }; + let Some(mut position) = cache.position_owned(&position_id) else { + return Ok(false); + }; + + if !position.is_opposite_side(fill.order_side) || fill.last_qty.raw <= position.quantity.raw { + return Ok(false); + } + + if position.side != PositionSide::Flat && position.trade_ids().contains(&fill.trade_id) { + return Ok(true); + } + + if !context.allow_deferred_orderless_flips { + return Err(apply_error( + entry, + "orderless position flip requires snapshot-tail replay context to match the following PositionOpened event", + )); + } + + let oms_type = cache.oms_type(&position_id).unwrap_or(OmsType::Unspecified); + let (closing_fill, opening_fill) = fill + .split_for_position_flip(position.quantity, None, fill.event_id) + .map_err(|e| apply_error(entry, e))?; + position.apply(&closing_fill); + apply_result(entry, cache.update_position(&position))?; + context.push_orderless_flip(entry.seq, position_id, oms_type, opening_fill); + Ok(true) +} + +fn apply_pending_orderless_flip_opened( + cache: &mut Cache, + entry: &EventStoreEntry, + opened: &PositionOpened, + context: &mut CacheReplayContext, +) -> Result, CacheReplayError> { + let Some((opening_fill, oms_type)) = context.take_opening_fill(entry, opened)? else { + return Ok(None); + }; + let instrument = cache + .instrument(&opening_fill.instrument_id) + .cloned() + .ok_or_else(|| { + apply_error( + entry, + format!( + "instrument {} not found for orderless flip position {}", + opening_fill.instrument_id, opened.position_id, + ), + ) + })?; + let prior = cache.position_owned(&opened.position_id); + let mut position = Position::new(&instrument, opening_fill); + if let Some(prior) = prior { + let current_replay = position.replay_events.clone(); + position.replay_events = prior.replay_events; + position.replay_events.extend(current_replay); + position.fill_voids = prior.fill_voids; + } + apply_result(entry, cache.add_position_without_order(&position, oms_type))?; + + apply_position_opened(cache, entry, opened).map(Some) +} + // Returns `Ok(true)` when the position side applied (no position association, or an // idempotent replay no-op) and `Ok(false)` when the instrument needed to open the // position is missing, so the report's ignored count surfaces the divergence. @@ -1302,6 +1528,7 @@ fn apply_fill_to_position( cache: &mut Cache, entry: &EventStoreEntry, fill: &OrderFilled, + orderless_leg_fill: bool, ) -> Result { let Some(position_id) = fill.position_id else { return Ok(true); @@ -1330,10 +1557,34 @@ fn apply_fill_to_position( }; let position = Position::new(&instrument, fill.clone()); - apply_result(entry, cache.add_position(&position, OmsType::Unspecified))?; + + if orderless_leg_fill { + apply_result( + entry, + cache.add_position_without_order(&position, OmsType::Unspecified), + )?; + } else { + apply_result(entry, cache.add_position(&position, OmsType::Unspecified))?; + } Ok(true) } +fn is_orderless_leg_fill(cache: &Cache, fill: &OrderFilled) -> bool { + if !fill.client_order_id.as_str().contains("-LEG-") + && !fill.venue_order_id.as_str().contains("-LEG-") + { + return false; + } + + let is_non_spread_instrument = cache + .instrument(&fill.instrument_id) + .is_none_or(|instrument| !instrument.is_spread()); + + is_non_spread_instrument + && !cache.order_exists(&fill.client_order_id) + && cache.client_order_id(&fill.venue_order_id).is_none() +} + fn apply_fill_void_to_order_and_positions( cache: &mut Cache, entry: &EventStoreEntry, @@ -2232,6 +2483,11 @@ mod tests { CacheMutationRecoveryClass::EventStoreCapturedAndReplayed, &[PAYLOAD_TYPE_ORDER_FILLED], ), + cache_mutation( + "add_position_without_order", + CacheMutationRecoveryClass::EventStoreCapturedAndReplayed, + &[PAYLOAD_TYPE_ORDER_FILLED], + ), cache_mutation( "update_account", CacheMutationRecoveryClass::EventStoreCapturedAndReplayed, @@ -3446,6 +3702,228 @@ mod tests { assert_eq!(position.commissions(), vec![Money::from("1 USD")]); } + #[rstest] + fn orderless_leg_fill_replay_creates_position_without_order_mapping() { + let instrument = InstrumentAny::CurrencyPair(audusd_sim()); + let client_order_id = ClientOrderId::from("SPREAD-LEG-AUDUSD"); + let position_id = PositionId::from("P-ORDERLESS-LEG"); + let filled = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(client_order_id) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-AUDUSD")) + .position_id(position_id) + .commission(Money::from("1 USD")) + .build(); + let reader = reader_with_entries( + "run-orderless-leg-fill-replay", + &[append_order_event( + 1, + &OrderEventAny::Filled(filled.clone()), + )], + ); + let mut cache = Cache::default(); + cache.add_instrument(instrument).expect("add instrument"); + + let report = replay_cache_snapshot_tail(&mut cache, &reader).expect("replay"); + let position = cache + .position_owned(&position_id) + .expect("orderless leg position replayed"); + + assert_eq!(report.applied_entries, 1); + assert_eq!(report.ignored_entries, 0); + assert!(cache.order_owned(&client_order_id).is_none()); + assert_eq!(cache.position_id(&client_order_id), None); + assert_eq!(position.event_count(), 1); + assert_eq!(position.last_event(), Some(filled.clone())); + assert_eq!(position.trade_ids(), vec![filled.trade_id]); + assert_eq!(position.commissions(), vec![Money::from("1 USD")]); + assert!(cache.check_integrity()); + } + + #[rstest] + fn orderless_hedging_flip_replay_recreates_replacement_position() { + let instrument = InstrumentAny::CurrencyPair(audusd_sim()); + let client_order_id = ClientOrderId::from("SPREAD-LEG-AUDUSD"); + let first_position_id = PositionId::from("P-ORDERLESS-LEG-1"); + let replacement_position_id = PositionId::from("P-ORDERLESS-LEG-2"); + let opening_fill = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(client_order_id) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-1")) + .trade_id(TradeId::from("T-SPREAD-LEG-1")) + .order_side(OrderSide::Buy) + .last_qty(Quantity::from(1)) + .last_px(Price::from("1.00000")) + .position_id(first_position_id) + .commission(Money::from("1 USD")) + .build(); + let first_position = Position::new(&instrument, opening_fill.clone()); + let first_opened = PositionOpened::create( + &first_position, + &opening_fill, + UUID4::new(), + opening_fill.ts_init, + ); + + let flip_fill = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(client_order_id) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-2")) + .trade_id(TradeId::from("T-SPREAD-LEG-2")) + .order_side(OrderSide::Sell) + .last_qty(Quantity::from(2)) + .last_px(Price::from("1.10000")) + .position_id(first_position_id) + .commission(Money::from("2 USD")) + .build(); + let mut closing_fragment = flip_fill.clone(); + closing_fragment.last_qty = Quantity::from(1); + closing_fragment.commission = Some(Money::from("1 USD")); + let mut closed_position = first_position; + closed_position.apply(&closing_fragment); + let first_closed = PositionClosed::create( + &closed_position, + &closing_fragment, + UUID4::new(), + flip_fill.ts_init, + ); + + let mut opening_fragment = flip_fill.clone(); + opening_fragment.last_qty = Quantity::from(1); + opening_fragment.position_id = Some(replacement_position_id); + opening_fragment.commission = Some(Money::from("1 USD")); + opening_fragment.event_id = UUID4::new(); + opening_fragment.causation_id = Some(flip_fill.event_id); + let mut replacement_position = Position::new(&instrument, opening_fragment.clone()); + let replacement_opened = PositionOpened::create( + &replacement_position, + &opening_fragment, + UUID4::new(), + opening_fragment.ts_init, + ); + + let subsequent_fill = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(client_order_id) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-3")) + .trade_id(TradeId::from("T-SPREAD-LEG-3")) + .order_side(OrderSide::Sell) + .last_qty(Quantity::from(1)) + .last_px(Price::from("1.20000")) + .position_id(replacement_position_id) + .commission(Money::from("1 USD")) + .build(); + replacement_position.apply(&subsequent_fill); + let replacement_changed = PositionChanged::create( + &replacement_position, + &subsequent_fill, + UUID4::new(), + subsequent_fill.ts_init, + ); + let reader = reader_with_entries( + "run-orderless-hedging-flip-replay", + &[ + append_order_event(1, &OrderEventAny::Filled(opening_fill)), + append_position_event(2, &PositionEvent::PositionOpened(first_opened)), + append_order_event(3, &OrderEventAny::Filled(flip_fill.clone())), + append_position_event(4, &PositionEvent::PositionClosed(first_closed)), + append_position_event(5, &PositionEvent::PositionOpened(replacement_opened)), + append_order_event(6, &OrderEventAny::Filled(subsequent_fill.clone())), + append_position_event(7, &PositionEvent::PositionChanged(replacement_changed)), + ], + ); + let mut cache = Cache::default(); + cache.add_instrument(instrument).expect("add instrument"); + + let report = replay_cache_snapshot_tail(&mut cache, &reader).expect("replay"); + + assert_eq!(report.applied_entries, 7); + assert_eq!(report.ignored_entries, 0); + let closed = cache + .position_owned(&first_position_id) + .expect("closed predecessor replayed"); + assert!(closed.is_closed()); + assert_eq!(closed.event_count(), 2); + let closing_fragments = closed.fill_fragments(client_order_id, flip_fill.trade_id); + assert_eq!(closing_fragments.len(), 1); + assert_eq!(closing_fragments[0].last_qty, Quantity::from(1)); + assert_eq!(closing_fragments[0].commission, Some(Money::from("1 USD"))); + assert_eq!(closing_fragments[0].event_id, flip_fill.event_id); + + let replacement = cache + .position_owned(&replacement_position_id) + .expect("open replacement replayed"); + assert!(replacement.is_open()); + assert_eq!(replacement.side, PositionSide::Short); + assert_eq!(replacement.quantity, Quantity::from(2)); + assert_eq!(replacement.event_count(), 2); + assert_eq!( + cache.oms_type(&replacement_position_id), + Some(OmsType::Hedging) + ); + assert!(replacement.trade_ids().contains(&flip_fill.trade_id)); + assert!(replacement.trade_ids().contains(&subsequent_fill.trade_id)); + let opening_fragments = replacement.fill_fragments(client_order_id, flip_fill.trade_id); + assert_eq!(opening_fragments.len(), 1); + assert_eq!(opening_fragments[0].last_qty, Quantity::from(1)); + assert_eq!(opening_fragments[0].commission, Some(Money::from("1 USD"))); + assert_eq!(opening_fragments[0].causation_id, Some(flip_fill.event_id)); + + assert!(cache.orders_for_position(&first_position_id).is_empty()); + assert!( + cache + .orders_for_position(&replacement_position_id) + .is_empty() + ); + assert_eq!(cache.position_id(&client_order_id), None); + assert!(cache.check_integrity()); + } + + #[rstest] + fn single_entry_orderless_flip_is_rejected_before_mutating_position() { + let instrument = InstrumentAny::CurrencyPair(audusd_sim()); + let position_id = PositionId::from("P-ORDERLESS-SINGLE-ENTRY"); + let opening_fill = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(ClientOrderId::from("SPREAD-LEG-SINGLE")) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-SINGLE-1")) + .trade_id(TradeId::from("T-SPREAD-LEG-SINGLE-1")) + .order_side(OrderSide::Buy) + .last_qty(Quantity::from(1)) + .position_id(position_id) + .build(); + let original = Position::new(&instrument, opening_fill.clone()); + let flip_fill = OrderFilledSpec::builder() + .instrument_id(instrument.id()) + .client_order_id(opening_fill.client_order_id) + .venue_order_id(VenueOrderId::from("V-SPREAD-LEG-SINGLE-2")) + .trade_id(TradeId::from("T-SPREAD-LEG-SINGLE-2")) + .order_side(OrderSide::Sell) + .last_qty(Quantity::from(2)) + .position_id(position_id) + .build(); + let entry = append_order_event(1, &OrderEventAny::Filled(flip_fill)).entry; + let mut cache = Cache::default(); + cache + .add_instrument(instrument) + .expect("add replay instrument"); + cache + .add_position_without_order(&original, OmsType::Hedging) + .expect("seed orderless position"); + + let error = apply_cache_replay_entry(&mut cache, &entry) + .expect_err("single-entry API cannot defer the opening fragment"); + let after = cache + .position_owned(&position_id) + .expect("position retained"); + + assert!(error.to_string().contains("snapshot-tail replay context")); + assert_eq!(after.side, original.side); + assert_eq!(after.quantity, original.quantity); + assert_eq!(after.event_count(), original.event_count()); + assert_eq!(after.trade_ids(), original.trade_ids()); + } + #[rstest] fn order_fill_replay_without_instrument_counts_fill_as_ignored() { // The position side cannot open without the instrument; the fill must count @@ -3829,7 +4307,7 @@ mod tests { .add_position(&position, OmsType::Unspecified) .expect("seed position"); - let applied = apply_fill_to_position(&mut cache, &entry, &fill).expect("apply fill"); + let applied = apply_fill_to_position(&mut cache, &entry, &fill, false).expect("apply fill"); let position = cache .position_owned(&position_id) .expect("position updated"); @@ -3880,7 +4358,7 @@ mod tests { .add_position(&position, OmsType::Unspecified) .expect("seed position"); - let applied = apply_fill_to_position(&mut cache, &entry, &dup_fill).expect("apply"); + let applied = apply_fill_to_position(&mut cache, &entry, &dup_fill, false).expect("apply"); let position = cache .position_owned(&position_id) .expect("position updated"); @@ -3906,7 +4384,7 @@ mod tests { let entry = append_order_event(1, &OrderEventAny::Filled(fill.clone())).entry; let mut cache = Cache::default(); - let applied = apply_fill_to_position(&mut cache, &entry, &fill).expect("apply"); + let applied = apply_fill_to_position(&mut cache, &entry, &fill, false).expect("apply"); assert!( !applied, diff --git a/crates/execution/src/engine/mod.rs b/crates/execution/src/engine/mod.rs index de8f903cb37f..2945e1b75e03 100644 --- a/crates/execution/src/engine/mod.rs +++ b/crates/execution/src/engine/mod.rs @@ -2782,14 +2782,54 @@ impl ExecutionEngine { } match oms_type { - OmsType::Hedging => fill - .position_id + OmsType::Hedging => self + .orderless_hedging_leg_position_id(fill) + .or(fill.position_id) .unwrap_or_else(|| self.pos_id_generator.generate(fill.strategy_id, false)), OmsType::Netting => self.determine_netting_position_id(fill), _ => self.determine_netting_position_id(fill), } } + fn orderless_hedging_leg_position_id(&self, fill: &OrderFilled) -> Option { + if !self.is_leg_fill(fill) { + return None; + } + + let cache = self.cache.borrow(); + if cache.order_exists(&fill.client_order_id()) { + return None; + } + + let matching_positions: Vec = cache + .positions_open( + Some(&fill.instrument_id.venue), + Some(&fill.instrument_id), + Some(&fill.strategy_id), + Some(&fill.account_id), + None, + ) + .iter() + .filter(|position| position.opening_order_id == fill.client_order_id) + .map(|position| position.id) + .collect(); + + match matching_positions.as_slice() { + [position_id] => Some(*position_id), + [] => None, + _ => { + log::warn!( + "Cannot uniquely correlate HEDGING leg fill {} to an orderless position: \ + found {} positions with opening_order_id={}", + fill.trade_id, + matching_positions.len(), + fill.client_order_id, + ); + None + } + } + } + fn is_leg_fill(&self, fill: &OrderFilled) -> bool { if !fill.client_order_id.as_str().contains("-LEG-") && !fill.venue_order_id.as_str().contains("-LEG-") @@ -3766,7 +3806,15 @@ impl ExecutionEngine { position.replay_events.extend(current_replay); position.fill_voids = prior.fill_voids; } - self.cache.borrow_mut().add_position(&position, oms_type)?; + let is_orderless_leg = self.is_leg_fill(&fill) + && !self.cache.borrow().order_exists(&position.opening_order_id); + if is_orderless_leg { + self.cache + .borrow_mut() + .add_position_without_order(&position, oms_type)?; + } else { + self.cache.borrow_mut().add_position(&position, oms_type)?; + } if self.config.snapshot_positions { self.create_position_state_snapshot(&position, true); @@ -3901,82 +3949,11 @@ impl ExecutionEngine { oms_type: OmsType, ) -> Vec { let mut position_events = Vec::new(); - let difference = match position.side { - PositionSide::Long => Quantity::from_raw( - fill.last_qty.raw - position.quantity.raw, - position.size_precision, - ), - PositionSide::Short => Quantity::from_raw( - position.quantity.raw.abs_diff(fill.last_qty.raw), // Equivalent to Python's abs(position.quantity - fill.last_qty) - position.size_precision, - ), - _ => fill.last_qty, - }; - // Split commission between two positions - let fill_percent = position.quantity.as_decimal() / fill.last_qty.as_decimal(); - let (commission1, commission2) = if let Some(commission) = fill.commission { - let commission_currency = commission.currency; - let commission1 = - Money::from_decimal(commission.as_decimal() * fill_percent, commission_currency) - .expect("Invalid split commission"); - let commission2 = commission - commission1; - (Some(commission1), Some(commission2)) - } else { + if fill.commission.is_none() { log::warn!( "Commission is not available for position flip, splitting with no commission" ); - (None, None) - }; - - let mut fill_split1: Option = None; - - if position.is_open() { - let mut split = OrderFilled::new( - fill.trader_id, - fill.strategy_id, - fill.instrument_id, - fill.client_order_id, - fill.venue_order_id, - fill.account_id, - fill.trade_id, - fill.order_side, - fill.order_type, - position.quantity, - fill.last_px, - fill.currency, - fill.liquidity_side, - fill.event_id, - fill.ts_event, - fill.ts_init, - fill.reconciliation, - fill.position_id, - commission1, - fill.info.clone(), - ); - split.causation_id = fill.causation_id; - fill_split1 = Some(split); - - if let Some(position_event) = - self.update_position(position, fill_split1.as_ref().unwrap()) - { - position_events.push(position_event); - } - - // Snapshot closed position before reusing ID (NETTING mode) - if oms_type == OmsType::Netting - && let Err(e) = self.snapshot_position(position) - { - log::warn!("Failed to snapshot position during flip: {e:?}"); - } - } - - // Guard against flipping a position with a zero fill size - if difference.raw == 0 { - log::warn!( - "Zero fill size during position flip calculation, this could be caused by a mismatch between instrument `size_precision` and a quantity `size_precision`" - ); - return position_events; } let position_id_flip = if oms_type == OmsType::Hedging @@ -3990,29 +3967,20 @@ impl ExecutionEngine { fill.position_id }; - let mut fill_split2 = OrderFilled::new( - fill.trader_id, - fill.strategy_id, - fill.instrument_id, - fill.client_order_id, - fill.venue_order_id, - fill.account_id, - fill.trade_id, - fill.order_side, - fill.order_type, - difference, - fill.last_px, - fill.currency, - fill.liquidity_side, - UUID4::new(), - fill.ts_event, - fill.ts_init, - fill.reconciliation, - position_id_flip, - commission2, - fill.info.clone(), - ); - fill_split2.causation_id = Some(fill.event_id); + let (fill_split1, fill_split2) = fill + .split_for_position_flip(position.quantity, position_id_flip, UUID4::new()) + .expect("Invalid position flip split"); + + if let Some(position_event) = self.update_position(position, &fill_split1) { + position_events.push(position_event); + } + + // Snapshot closed position before reusing ID (NETTING mode) + if oms_type == OmsType::Netting + && let Err(e) = self.snapshot_position(position) + { + log::warn!("Failed to snapshot position during flip: {e:?}"); + } if oms_type == OmsType::Hedging && let Some(position_id) = fill.position_id diff --git a/crates/execution/tests/exec_engine.rs b/crates/execution/tests/exec_engine.rs index cc6aa558a62e..033996443a5d 100644 --- a/crates/execution/tests/exec_engine.rs +++ b/crates/execution/tests/exec_engine.rs @@ -2433,10 +2433,34 @@ fn test_process_leg_fill_without_order_updates_position_and_publishes_order_befo assert_eq!(position.id, expected_position_id); assert_eq!(position.opening_order_id, client_order_id); assert_eq!(position.quantity, Quantity::from(1)); + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); assert_eq!( topics.borrow().as_slice(), ["portfolio", "orders", "positions"] ); + drop(position); + drop(cache); + let mut cache = execution_engine.cache().borrow_mut(); + assert!(cache.check_integrity()); + + cache.clear_index(); + cache.build_index(); + let instrument_id = instrument.id(); + assert!(cache.position(&expected_position_id).is_some()); + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); + assert_eq!( + cache.positions_open_count( + Some(&instrument_id.venue), + Some(&instrument_id), + Some(&strategy_id), + None, + None, + ), + 1, + ); + assert!(cache.check_integrity()); } #[rstest] @@ -2480,6 +2504,133 @@ fn test_process_duplicate_leg_fill_without_order_does_not_reapply_position( ); } +#[rstest] +fn test_hedging_leg_fill_without_order_reuses_position_for_same_synthetic_leg( + mut execution_engine: ExecutionEngine, +) { + *msgbus::get_message_bus().borrow_mut() = MessageBus::default(); + + let (_instrument, fill, _) = prepare_leg_fill_without_order(&execution_engine); + let strategy_id = fill.strategy_id; + let client_order_id = fill.client_order_id; + execution_engine.register_oms_type(strategy_id, OmsType::Hedging); + + execution_engine.process(&OrderEventAny::Filled(fill.clone())); + let position_id = { + let cache = execution_engine.cache().borrow(); + let positions = cache.positions_open( + Some(&fill.instrument_id.venue), + Some(&fill.instrument_id), + Some(&strategy_id), + Some(&fill.account_id), + None, + ); + + assert_eq!(positions.len(), 1); + assert_eq!(positions[0].opening_order_id, client_order_id); + assert_eq!(positions[0].quantity, Quantity::from(1)); + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); + positions[0].id + }; + + let mut second_fill = fill.clone(); + second_fill.trade_id = TradeId::new("T-LEG-002"); + second_fill.venue_order_id = VenueOrderId::from("V-SPREAD-LEG-2"); + second_fill.position_id = None; + execution_engine.process(&OrderEventAny::Filled(second_fill.clone())); + + let mut cache = execution_engine.cache().borrow_mut(); + assert_eq!(cache.positions_total_count(None, None, None, None, None), 1); + { + let position = cache + .position(&position_id) + .expect("second leg fill should update the first HEDGING position"); + assert_eq!(position.quantity, Quantity::from(2)); + assert_eq!(position.event_count(), 2); + assert!(position.trade_ids.contains(&fill.trade_id)); + assert!(position.trade_ids.contains(&second_fill.trade_id)); + assert_eq!(position.opening_order_id, client_order_id); + } + assert!(!cache.order_exists(&client_order_id)); + assert_eq!(cache.position_id(&client_order_id), None); + assert!(cache.check_integrity()); +} + +#[rstest] +fn test_hedging_leg_fill_without_order_reuses_open_position_after_flip( + mut execution_engine: ExecutionEngine, +) { + *msgbus::get_message_bus().borrow_mut() = MessageBus::default(); + + let (_instrument, fill, _) = prepare_leg_fill_without_order(&execution_engine); + let strategy_id = fill.strategy_id; + let client_order_id = fill.client_order_id; + execution_engine.register_oms_type(strategy_id, OmsType::Hedging); + + execution_engine.process(&OrderEventAny::Filled(fill.clone())); + + let mut flip_fill = fill.clone(); + flip_fill.trade_id = TradeId::new("T-LEG-002"); + flip_fill.venue_order_id = VenueOrderId::from("V-SPREAD-LEG-2"); + flip_fill.order_side = OrderSide::Sell; + flip_fill.last_qty = Quantity::from(2); + flip_fill.position_id = None; + execution_engine.process(&OrderEventAny::Filled(flip_fill.clone())); + + let flipped_position_id = { + let cache = execution_engine.cache().borrow(); + let open_positions = cache.positions_open( + Some(&fill.instrument_id.venue), + Some(&fill.instrument_id), + Some(&strategy_id), + Some(&fill.account_id), + None, + ); + let closed_positions = cache.positions_closed( + Some(&fill.instrument_id.venue), + Some(&fill.instrument_id), + Some(&strategy_id), + Some(&fill.account_id), + None, + ); + + assert_eq!(open_positions.len(), 1); + assert_eq!(closed_positions.len(), 1); + assert_eq!(open_positions[0].opening_order_id, client_order_id); + assert_eq!(closed_positions[0].opening_order_id, client_order_id); + assert_eq!(open_positions[0].side, PositionSide::Short); + assert_eq!(open_positions[0].quantity, Quantity::from(1)); + open_positions[0].id + }; + + let mut subsequent_fill = fill; + subsequent_fill.trade_id = TradeId::new("T-LEG-003"); + subsequent_fill.venue_order_id = VenueOrderId::from("V-SPREAD-LEG-3"); + subsequent_fill.order_side = OrderSide::Sell; + subsequent_fill.position_id = None; + execution_engine.process(&OrderEventAny::Filled(subsequent_fill.clone())); + + let mut cache = execution_engine.cache().borrow_mut(); + assert_eq!(cache.positions_total_count(None, None, None, None, None), 2); + assert_eq!(cache.positions_open_count(None, None, None, None, None), 1); + assert_eq!( + cache.positions_closed_count(None, None, None, None, None), + 1 + ); + { + let position = cache + .position(&flipped_position_id) + .expect("subsequent leg fill should update the open flipped position"); + assert_eq!(position.side, PositionSide::Short); + assert_eq!(position.quantity, Quantity::from(2)); + assert_eq!(position.event_count(), 2); + assert!(position.trade_ids.contains(&flip_fill.trade_id)); + assert!(position.trade_ids.contains(&subsequent_fill.trade_id)); + } + assert!(cache.check_integrity()); +} + #[rstest] fn test_project_reconciliation_fill_applies_no_portfolio_economics_on_cash_account( mut execution_engine: ExecutionEngine, diff --git a/crates/model/src/events/order/filled.rs b/crates/model/src/events/order/filled.rs index d61ec95bbc9f..859101475eb7 100644 --- a/crates/model/src/events/order/filled.rs +++ b/crates/model/src/events/order/filled.rs @@ -155,6 +155,54 @@ impl OrderFilled { pub fn is_sell(&self) -> bool { self.order_side == OrderSide::Sell } + + /// Splits an overfill into the fragment which closes the current position and the + /// fragment which opens the flipped position. + /// + /// # Errors + /// + /// Returns an error when `closing_qty` is zero, is not smaller than the fill quantity, + /// or the proportional commission cannot be represented. + pub fn split_for_position_flip( + &self, + closing_qty: Quantity, + opening_position_id: Option, + opening_event_id: UUID4, + ) -> anyhow::Result<(Self, Self)> { + anyhow::ensure!(!closing_qty.is_zero(), "closing quantity was zero"); + anyhow::ensure!( + closing_qty.raw < self.last_qty.raw, + "closing quantity {closing_qty} must be smaller than fill quantity {}", + self.last_qty, + ); + + let opening_qty = + Quantity::from_raw(self.last_qty.raw - closing_qty.raw, closing_qty.precision); + let closing_fraction = closing_qty.as_decimal() / self.last_qty.as_decimal(); + let (closing_commission, opening_commission) = match self.commission { + Some(commission) => { + let closing = Money::from_decimal( + commission.as_decimal() * closing_fraction, + commission.currency, + )?; + (Some(closing), Some(commission - closing)) + } + None => (None, None), + }; + + let mut closing = self.clone(); + closing.last_qty = closing_qty; + closing.commission = closing_commission; + + let mut opening = self.clone(); + opening.last_qty = opening_qty; + opening.position_id = opening_position_id; + opening.commission = opening_commission; + opening.event_id = opening_event_id; + opening.causation_id = Some(self.event_id); + + Ok((closing, opening)) + } } impl Debug for OrderFilled { @@ -504,6 +552,35 @@ mod tests { assert!(!order_filled.is_buy()); } + #[rstest] + fn test_split_for_position_flip_preserves_provenance_and_commission() { + let mut fill = create_test_order_filled(); + fill.last_qty = Quantity::from(100); + fill.commission = Some(Money::new(2.5, Currency::USD())); + let source_event_id = fill.event_id; + let opening_event_id = UUID4::new(); + let opening_position_id = PositionId::from("P-FLIPPED"); + + let (closing, opening) = fill + .split_for_position_flip( + Quantity::from(40), + Some(opening_position_id), + opening_event_id, + ) + .expect("split fill"); + + assert_eq!(closing.last_qty, Quantity::from(40)); + assert_eq!(closing.position_id, fill.position_id); + assert_eq!(closing.event_id, source_event_id); + assert_eq!(closing.causation_id, fill.causation_id); + assert_eq!(closing.commission, Some(Money::new(1.0, Currency::USD()))); + assert_eq!(opening.last_qty, Quantity::from(60)); + assert_eq!(opening.position_id, Some(opening_position_id)); + assert_eq!(opening.event_id, opening_event_id); + assert_eq!(opening.causation_id, Some(source_event_id)); + assert_eq!(opening.commission, Some(Money::new(1.5, Currency::USD()))); + } + #[rstest] fn test_order_filled_specified_side() { let buy_order = create_test_order_filled();