nautilus-common 0.57.0

// -------------------------------------------------------------------------------------------------
//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
//  https://nautechsystems.io
//
//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
//  You may not use this file except in compliance with the License.
//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
// -------------------------------------------------------------------------------------------------

//! Tests module for `Cache`.

#[cfg(feature = "defi")]
use std::sync::Arc;
use std::{borrow::Cow, cell::RefCell, rc::Rc};

use ahash::{AHashMap, AHashSet};
use bytes::Bytes;
use nautilus_core::{UUID4, UnixNanos};
#[cfg(feature = "defi")]
use nautilus_model::defi::{
    AmmType, Dex, DexType, Pool, PoolIdentifier, PoolProfiler, Token, chain::chains,
};
use nautilus_model::{
    accounts::AccountAny,
    data::{
        Bar, BarType, CustomData, DataType, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus,
        MarkPriceUpdate, QuoteTick, TradeTick,
    },
    enums::{
        AccountType, AggressorSide, AssetClass, BookType, ContingencyType, InstrumentClass,
        LiquiditySide, MarketStatusAction, OmsType, OptionKind, OrderSide, OrderStatus, OrderType,
        PositionSide, PriceType, TimeInForce, TriggerType,
    },
    events::{
        AccountState, OrderAccepted, OrderCanceled, OrderEmulated, OrderEventAny, OrderFilled,
        OrderRejected, OrderReleased, OrderSnapshot, OrderSubmitted, OrderUpdated,
        position::snapshot::PositionSnapshot,
    },
    identifiers::{
        AccountId, ClientId, ClientOrderId, ComponentId, ExecAlgorithmId, InstrumentId,
        OrderListId, PositionId, StrategyId, Symbol, TradeId, Venue, VenueOrderId,
    },
    instruments::{
        CurrencyPair, Instrument, InstrumentAny, OptionContract, SyntheticInstrument, stubs::*,
    },
    orderbook::OrderBook,
    orders::{
        Order, OrderAny, OrderError, OrderList,
        builder::OrderTestBuilder,
        stubs::{TestOrderEventStubs, TestOrdersGenerator},
    },
    position::Position,
    stubs::TestDefault,
    types::{AccountBalance, Currency, Money, Price, Quantity},
};
use rstest::{fixture, rstest};
use ustr::Ustr;

use crate::{
    cache::{
        Cache, CacheConfig, CacheView, OrderRef,
        database::{CacheDatabaseAdapter, CacheMap},
    },
    signal::Signal,
};

#[fixture]
fn cache() -> Cache {
    Cache::default()
}

fn assert_orders_eq(actual: &[OrderRef<'_>], expected: &[&OrderAny]) {
    assert_eq!(actual.len(), expected.len(), "order list length mismatch");
    for (a, e) in actual.iter().zip(expected.iter().copied()) {
        assert_eq!(a, e);
    }
}

fn orders_contains(actual: &[OrderRef<'_>], expected: &OrderAny) -> bool {
    actual.iter().any(|r| r == expected)
}

#[rstest]
fn test_cache_view_borrows_same_cache(audusd_sim: CurrencyPair) {
    let cache = Rc::new(RefCell::new(Cache::default()));
    cache
        .borrow_mut()
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let view = CacheView::from(cache);
    let borrowed = view.borrow();

    assert!(borrowed.instrument(&audusd_sim.id).is_some());
}

#[rstest]
#[should_panic]
fn test_cache_view_borrow_panics_when_mutably_borrowed() {
    let cache = Rc::new(RefCell::new(Cache::default()));
    let view = CacheView::from(cache.clone());
    let _mutable_borrow = cache.borrow_mut();

    let _borrowed = view.borrow();
}

#[rstest]
fn test_build_index_when_empty(mut cache: Cache) {
    cache.build_index();
}

#[rstest]
fn test_check_integrity_when_empty(mut cache: Cache) {
    let result = cache.check_integrity();
    assert!(result);
}

#[rstest]
fn test_check_residuals_when_empty(cache: Cache) {
    let result = cache.check_residuals();
    assert!(!result);
}

#[rstest]
fn test_clear_index_when_empty(mut cache: Cache) {
    cache.clear_index();
}

#[rstest]
fn test_reset_when_empty(mut cache: Cache) {
    cache.reset();
}

#[rstest]
#[case(true, false)]
#[case(false, true)]
fn test_reset_honors_drop_instruments_on_reset(
    audusd_sim: CurrencyPair,
    #[case] drop_on_reset: bool,
    #[case] retained: bool,
) {
    let config = CacheConfig::builder()
        .drop_instruments_on_reset(drop_on_reset)
        .build();
    let mut cache = Cache::new(Some(config), None);

    let instrument = InstrumentAny::CurrencyPair(audusd_sim.clone());
    cache.add_instrument(instrument).unwrap();
    assert!(cache.instrument(&audusd_sim.id).is_some());

    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    cache.add_order(order, None, None, false).unwrap();
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);

    cache.reset();

    assert_eq!(cache.orders_total_count(None, None, None, None, None), 0);
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 0);
    assert_eq!(cache.instrument(&audusd_sim.id).is_some(), retained);
}

#[rstest]
fn test_dispose_when_empty(mut cache: Cache) {
    cache.dispose();
}

#[rstest]
fn test_flush_db_when_empty(mut cache: Cache) {
    cache.flush_db();
}

#[rstest]
fn test_cache_general_when_no_database(mut cache: Cache) {
    assert!(cache.cache_general().is_ok());
}

#[rstest]
fn test_has_backing_reflects_injected_adapter() {
    let without = Cache::new(None, None);
    assert!(!without.has_backing());

    let with = Cache::new(None, Some(Box::new(SnapshotBlobTestDatabase::default())));
    assert!(with.has_backing());
}

#[rstest]
fn test_has_backing_after_set_database() {
    let mut cache = Cache::default();
    assert!(!cache.has_backing());

    cache.set_database(Box::new(SnapshotBlobTestDatabase::default()));
    assert!(cache.has_backing());
}

// -- EXECUTION -------------------------------------------------------------------------------

#[rstest]
fn test_cache_orders_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_orders()).is_ok());
}

#[rstest]
fn test_assign_position_ids_to_contingencies_propagates_parent_to_children(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let position_id = PositionId::new("P-1");
    let parent_id = ClientOrderId::from("PARENT-1");
    let child_a_id = ClientOrderId::from("CHILD-A");
    let child_b_id = ClientOrderId::from("CHILD-B");

    let mut parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oto)
        .linked_order_ids(vec![child_a_id, child_b_id])
        .build();
    parent.set_position_id(Some(position_id));
    let parent_strategy_id = parent.strategy_id();
    cache.add_order(parent, None, None, false).unwrap();

    let child_a = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("1.10000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_a_id)
        .build();
    cache.add_order(child_a, None, None, false).unwrap();

    let child_b = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("0.90000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_b_id)
        .build();
    cache.add_order(child_b, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();

    assert_eq!(
        cache.order(&child_a_id).unwrap().position_id(),
        Some(position_id),
    );
    assert_eq!(
        cache.order(&child_b_id).unwrap().position_id(),
        Some(position_id),
    );
    assert_eq!(cache.position_id(&child_a_id), Some(&position_id));
    assert_eq!(cache.position_id(&child_b_id), Some(&position_id));
    assert_eq!(
        cache.strategy_id_for_position(&position_id),
        Some(&parent_strategy_id),
    );

    let position_order_ids: AHashSet<ClientOrderId> = cache
        .orders_for_position(&position_id)
        .iter()
        .map(|o| o.client_order_id())
        .collect();
    assert!(position_order_ids.contains(&child_a_id));
    assert!(position_order_ids.contains(&child_b_id));
}

#[rstest]
fn test_assign_position_ids_to_contingencies_skips_non_oto_parent(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let position_id = PositionId::new("P-1");
    let parent_id = ClientOrderId::from("PARENT-1");
    let child_id = ClientOrderId::from("CHILD-1");

    let mut parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oco)
        .linked_order_ids(vec![child_id])
        .build();
    parent.set_position_id(Some(position_id));
    cache.add_order(parent, None, None, false).unwrap();

    let child = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("1.10000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_id)
        .build();
    cache.add_order(child, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();

    assert_eq!(cache.order(&child_id).unwrap().position_id(), None);
    assert_eq!(cache.position_id(&child_id), None);
}

#[rstest]
fn test_assign_position_ids_to_contingencies_skips_when_parent_has_no_position_id(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let parent_id = ClientOrderId::from("PARENT-1");
    let child_id = ClientOrderId::from("CHILD-1");

    let parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oto)
        .linked_order_ids(vec![child_id])
        .build();
    cache.add_order(parent, None, None, false).unwrap();

    let child = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("1.10000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_id)
        .build();
    cache.add_order(child, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();

    assert_eq!(cache.order(&child_id).unwrap().position_id(), None);
    assert_eq!(cache.position_id(&child_id), None);
}

#[rstest]
fn test_assign_position_ids_to_contingencies_does_not_overwrite_existing_child_position_id(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let parent_position_id = PositionId::new("P-PARENT");
    let child_position_id = PositionId::new("P-CHILD");
    let parent_id = ClientOrderId::from("PARENT-1");
    let child_id = ClientOrderId::from("CHILD-1");

    let mut parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oto)
        .linked_order_ids(vec![child_id])
        .build();
    parent.set_position_id(Some(parent_position_id));
    cache.add_order(parent, None, None, false).unwrap();

    let mut child = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("1.10000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_id)
        .build();
    child.set_position_id(Some(child_position_id));
    cache.add_order(child, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();

    assert_eq!(
        cache.order(&child_id).unwrap().position_id(),
        Some(child_position_id),
    );
}

#[rstest]
fn test_assign_position_ids_to_contingencies_handles_missing_child(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let position_id = PositionId::new("P-1");
    let parent_id = ClientOrderId::from("PARENT-1");
    let absent_child_id = ClientOrderId::from("CHILD-ABSENT");

    let mut parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oto)
        .linked_order_ids(vec![absent_child_id])
        .build();
    parent.set_position_id(Some(position_id));
    cache.add_order(parent, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();

    assert_eq!(cache.position_id(&absent_child_id), None);
}

#[rstest]
fn test_assign_position_ids_to_contingencies_is_idempotent(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let position_id = PositionId::new("P-1");
    let parent_id = ClientOrderId::from("PARENT-1");
    let child_id = ClientOrderId::from("CHILD-1");

    let mut parent = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .client_order_id(parent_id)
        .contingency_type(ContingencyType::Oto)
        .linked_order_ids(vec![child_id])
        .build();
    parent.set_position_id(Some(position_id));
    cache.add_order(parent, None, None, false).unwrap();

    let child = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Sell)
        .price(Price::from("1.10000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(child_id)
        .build();
    cache.add_order(child, None, None, false).unwrap();

    cache.assign_position_ids_to_contingencies();
    cache.assign_position_ids_to_contingencies();

    assert_eq!(
        cache.order(&child_id).unwrap().position_id(),
        Some(position_id),
    );
    assert_eq!(cache.position_id(&child_id), Some(&position_id));
    assert_eq!(cache.orders_for_position(&position_id).len(), 1);
}

#[rstest]
fn test_order_when_empty(cache: Cache) {
    let client_order_id = ClientOrderId::test_default();
    let result = cache.order(&client_order_id);
    assert!(result.is_none());
}

#[rstest]
fn test_order_when_initialized(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();
    cache.add_order(order, None, None, false).unwrap();

    let order_ref = cache.order(&client_order_id).unwrap();
    let order = order_ref.clone();
    drop(order_ref);
    assert_orders_eq(&cache.orders(None, None, None, None, None), &[&order]);
    assert!(cache.orders_open(None, None, None, None, None).is_empty());
    assert!(cache.orders_closed(None, None, None, None, None).is_empty());
    assert_orders_eq(
        &cache.orders_active_local(None, None, None, None, None),
        &[&order],
    );
    assert!(
        cache
            .orders_emulated(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .orders_inflight(None, None, None, None, None)
            .is_empty()
    );
    assert!(cache.order_exists(&order.client_order_id()));
    assert!(!cache.is_order_open(&order.client_order_id()));
    assert!(!cache.is_order_closed(&order.client_order_id()));
    assert!(cache.is_order_active_local(&order.client_order_id()));
    assert!(!cache.is_order_emulated(&order.client_order_id()));
    assert!(!cache.is_order_inflight(&order.client_order_id()));
    assert!(!cache.is_order_pending_cancel_local(&order.client_order_id()));
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_closed_count(None, None, None, None, None), 0);
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        1
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_inflight_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
    assert_eq!(cache.venue_order_id(&order.client_order_id()), None);
}

#[rstest]
fn test_order_when_submitted(mut cache: Cache, audusd_sim: CurrencyPair) {
    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    // check the status change of the cached order
    let cached_order = cache.order(&client_order_id).unwrap();
    assert_eq!(cached_order.status(), OrderStatus::Submitted);

    let result = cache.order(&order.client_order_id()).unwrap();

    assert_eq!(order.status(), OrderStatus::Submitted);
    assert_eq!(&*result, &order);
    drop(result);
    assert_orders_eq(&cache.orders(None, None, None, None, None), &[&order]);
    assert!(cache.orders_open(None, None, None, None, None).is_empty());
    assert!(cache.orders_closed(None, None, None, None, None).is_empty());
    assert!(
        cache
            .orders_active_local(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .orders_emulated(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        !cache
            .orders_inflight(None, None, None, None, None)
            .is_empty()
    );
    assert!(cache.order_exists(&order.client_order_id()));
    assert!(!cache.is_order_open(&order.client_order_id()));
    assert!(!cache.is_order_closed(&order.client_order_id()));
    assert!(!cache.is_order_active_local(&order.client_order_id()));
    assert!(!cache.is_order_emulated(&order.client_order_id()));
    assert!(cache.is_order_inflight(&order.client_order_id()));
    assert!(!cache.is_order_pending_cancel_local(&order.client_order_id()));
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_closed_count(None, None, None, None, None), 0);
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        0
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_inflight_count(None, None, None, None, None), 1);
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
    assert_eq!(cache.venue_order_id(&order.client_order_id()), None);
}

#[rstest]
fn test_update_order_applies_event_to_cached_order(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();
    cache.add_order(order, None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    let applied = cache.update_order(&submitted).unwrap();
    let cached_order = cache.order(&client_order_id).unwrap();

    assert_eq!(applied.status(), OrderStatus::Submitted);
    assert_eq!(cached_order.status(), OrderStatus::Submitted);
    assert!(
        cache
            .orders_active_local(None, None, None, None, None)
            .is_empty()
    );
    assert_eq!(cache.orders_inflight(None, None, None, None, None).len(), 1);
    assert!(cache.is_order_inflight(&client_order_id));
}

#[rstest]
fn test_update_order_returns_not_found_for_missing_order(mut cache: Cache) {
    let event = OrderEventAny::Submitted(OrderSubmitted::default());
    let client_order_id = event.client_order_id();

    let err = cache.update_order(&event).unwrap_err();

    match err.downcast_ref::<OrderError>() {
        Some(OrderError::NotFound(id)) => assert_eq!(*id, client_order_id),
        other => panic!("Expected OrderError::NotFound, was {other:?}"),
    }
}

#[rstest]
fn test_update_order_rejects_invalid_transition_without_mutating(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order, None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    cache.update_order(&submitted).unwrap();
    let event_count = cache.order(&client_order_id).unwrap().event_count();

    let err = cache.update_order(&submitted).unwrap_err();
    let cached_order = cache.order(&client_order_id).unwrap();

    assert!(matches!(
        err.downcast_ref::<OrderError>(),
        Some(OrderError::InvalidStateTransition)
    ));
    assert_eq!(cached_order.status(), OrderStatus::Submitted);
    assert_eq!(
        cached_order.previous_status(),
        Some(OrderStatus::Initialized)
    );
    assert_eq!(cached_order.event_count(), event_count);
    assert!(cache.is_order_inflight(&client_order_id));
}

#[rstest]
fn test_order_mut_returns_none_for_missing_order(mut cache: Cache) {
    let client_order_id = ClientOrderId::from("O-MISSING");
    assert!(cache.order_mut(&client_order_id).is_none());
}

#[rstest]
fn test_order_owned_returns_independent_snapshot(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order, None, None, false).unwrap();

    let snapshot = cache.order_owned(&client_order_id).unwrap();

    // Snapshot is independent: subsequent cache mutations do not affect it.
    cache
        .order_mut(&client_order_id)
        .unwrap()
        .set_position_id(Some(PositionId::from("P-001")));

    assert!(snapshot.position_id().is_none());
    assert_eq!(
        cache.order(&client_order_id).unwrap().position_id(),
        Some(PositionId::from("P-001"))
    );
}

#[rstest]
fn test_order_mut_writes_propagate_to_subsequent_reads(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order, None, None, false).unwrap();

    cache
        .order_mut(&client_order_id)
        .unwrap()
        .set_position_id(Some(PositionId::from("P-001")));

    assert_eq!(
        cache.order(&client_order_id).unwrap().position_id(),
        Some(PositionId::from("P-001"))
    );
}

#[rstest]
fn test_add_order_replace_existing_overwrites_value(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let mut replacement = order;
    replacement.set_position_id(Some(PositionId::from("P-NEW")));
    cache.add_order(replacement, None, None, true).unwrap();

    assert_eq!(
        cache.order(&client_order_id).unwrap().position_id(),
        Some(PositionId::from("P-NEW"))
    );
}

#[rstest]
fn test_replace_order_overwrites_value(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let mut replacement = order;
    replacement.set_position_id(Some(PositionId::from("P-REPLACED")));
    cache.replace_order(&replacement).unwrap();

    assert_eq!(
        cache.order(&client_order_id).unwrap().position_id(),
        Some(PositionId::from("P-REPLACED"))
    );
}

#[rstest]
fn test_update_order_rejects_venue_fallback_when_event_client_id_differs(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let account_id = AccountId::from("SIM-001");
    let venue_order_id = VenueOrderId::from("V-001");
    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = TestOrderEventStubs::submitted(&order, account_id);
    update_order_with_event(&mut cache, &mut order, submitted);
    let accepted = TestOrderEventStubs::accepted(&order, account_id, venue_order_id);
    update_order_with_event(&mut cache, &mut order, accepted);
    let event_count = cache.order(&client_order_id).unwrap().event_count();

    let canceled = OrderEventAny::Canceled(OrderCanceled::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        ClientOrderId::from("UNKNOWN"),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        false,
        Some(venue_order_id),
        Some(account_id),
    ));

    let err = cache.update_order(&canceled).unwrap_err();
    let cached_order = cache.order(&client_order_id).unwrap();

    assert!(matches!(
        err.downcast_ref::<OrderError>(),
        Some(OrderError::Invariant(_))
    ));
    assert_eq!(cached_order.status(), OrderStatus::Accepted);
    assert_eq!(cached_order.event_count(), event_count);
    assert_eq!(
        cache.client_order_id(&venue_order_id),
        Some(&client_order_id)
    );
}

fn update_order_with_event(
    cache: &mut Cache,
    order: &mut OrderAny,
    event: impl Into<OrderEventAny>,
) {
    let event = event.into();
    *order = cache.update_order(&event).unwrap();
}

// Test order state transitions and cache queries when an order is rejected.
//
// This test verifies cache behavior for the complete lifecycle: initialized -> submitted -> rejected.
//
// PRODUCTION CODE BUG: This test fails at line 220 with:
//   assertion failed: cache.orders_emulated(None, None, None, None, None).is_empty()
//
// When an order transitions to REJECTED state, it incorrectly appears in the emulated orders
// collection. The cache should only track emulated orders separately, not include rejected orders.
//
// TODO: Fix cache order state management - rejected orders should not appear in emulated list.
// The bug is in production code (cache.rs), not in this test.
#[ignore = "Production bug: rejected orders incorrectly showing in emulated list"]
#[rstest]
fn test_order_when_rejected(mut cache: Cache, audusd_sim: CurrencyPair) {
    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    let rejected = OrderEventAny::Rejected(OrderRejected::default());
    update_order_with_event(&mut cache, &mut order, rejected);

    // check the status change of the cached order
    let cached_order = cache.order(&order.client_order_id()).unwrap();
    assert_eq!(cached_order.status(), OrderStatus::Rejected);

    let result = cache.order(&order.client_order_id()).unwrap();

    assert!(order.is_closed());
    assert_eq!(&*result, &order);
    drop(result);
    assert_orders_eq(&cache.orders(None, None, None, None, None), &[&order]);
    assert!(cache.orders_open(None, None, None, None, None).is_empty());
    assert_orders_eq(
        &cache.orders_closed(None, None, None, None, None),
        &[&order],
    );
    assert!(
        cache
            .orders_emulated(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .orders_inflight(None, None, None, None, None)
            .is_empty()
    );
    assert!(cache.order_exists(&order.client_order_id()));
    assert!(!cache.is_order_open(&order.client_order_id()));
    assert!(cache.is_order_closed(&order.client_order_id()));
    assert!(!cache.is_order_emulated(&order.client_order_id()));
    assert!(!cache.is_order_inflight(&order.client_order_id()));
    assert!(!cache.is_order_pending_cancel_local(&order.client_order_id()));
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_closed_count(None, None, None, None, None), 1);
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_inflight_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
}

#[rstest]
fn test_order_when_accepted(mut cache: Cache, audusd_sim: CurrencyPair) {
    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = OrderEventAny::Accepted(OrderAccepted::default());
    update_order_with_event(&mut cache, &mut order, accepted);

    let result = cache.order(&order.client_order_id()).unwrap();

    assert!(order.is_open());
    assert_eq!(&*result, &order);
    drop(result);
    assert_orders_eq(&cache.orders(None, None, None, None, None), &[&order]);
    assert_orders_eq(&cache.orders_open(None, None, None, None, None), &[&order]);
    assert!(cache.orders_closed(None, None, None, None, None).is_empty());
    assert!(
        cache
            .orders_emulated(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .orders_inflight(None, None, None, None, None)
            .is_empty()
    );
    assert!(cache.order_exists(&order.client_order_id()));
    assert!(cache.is_order_open(&order.client_order_id()));
    assert!(!cache.is_order_closed(&order.client_order_id()));
    assert!(!cache.is_order_emulated(&order.client_order_id()));
    assert!(!cache.is_order_inflight(&order.client_order_id()));
    assert!(!cache.is_order_pending_cancel_local(&order.client_order_id()));
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 1);
    assert_eq!(cache.orders_closed_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_inflight_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
    assert_eq!(
        cache.client_order_id(&order.venue_order_id().unwrap()),
        Some(&order.client_order_id())
    );
    assert_eq!(
        cache.venue_order_id(&order.client_order_id()),
        Some(&order.venue_order_id().unwrap())
    );
}

#[rstest]
fn test_client_order_ids_filtering(mut cache: Cache) {
    // Build a small deterministic universe: 2 venues × 3 instruments × 2 orders
    let venue_a = Venue::from("VENUE-A");
    let _venue_b = Venue::from("VENUE-B");

    let mut generator = TestOrdersGenerator::new(OrderType::Limit);
    generator.add_venue_and_total_instruments(venue_a, 3);
    generator.add_venue_and_total_instruments(_venue_b, 3);
    generator.set_orders_per_instrument(2);

    let orders = generator.build();

    let _instrument_a0 = InstrumentId::from("SYMBOL-0.VENUE-A");

    // Sanity-check the generated volume: 2 × 3 × 2 = 12
    assert_eq!(orders.len(), 12);

    // Load into cache so indices are built on the fly
    for order in &orders {
        cache.add_order(order.clone(), None, None, false).unwrap();
    }

    // No filters – expect all orders
    assert_eq!(
        cache.client_order_ids(None, None, None, None).len(),
        orders.len()
    );

    // Venue only
    let expected_venue_a = orders
        .iter()
        .filter(|o| o.instrument_id().venue == venue_a)
        .count();
    assert_eq!(
        cache
            .client_order_ids(Some(&venue_a), None, None, None)
            .len(),
        expected_venue_a
    );

    // Venue + instrument
    let instrument_a0 = InstrumentId::from("SYMBOL-0.VENUE-A");
    assert_eq!(
        cache
            .client_order_ids(Some(&venue_a), Some(&instrument_a0), None, None)
            .len(),
        orders
            .iter()
            .filter(|o| o.instrument_id() == instrument_a0)
            .count()
    );
}

#[rstest]
fn test_position_ids_filtering(mut cache: Cache) {
    fn make_pair(id_str: &str) -> CurrencyPair {
        CurrencyPair::new(
            InstrumentId::from(id_str),
            Symbol::from(id_str),
            Currency::USD(),
            Currency::EUR(),
            2,
            4,
            Price::from("0.01"),
            Quantity::from("0.0001"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    let venue_a = Venue::from("VENUE-A");
    let _venue_b = Venue::from("VENUE-B");

    // Build two open positions and one closed position across venues
    let instr_a0 = make_pair("PAIR-0.VENUE-A");
    let instr_b0 = make_pair("PAIR-0.VENUE-B");

    let base_order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_a0.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from("1"))
        .build();

    let fill_a_event = TestOrderEventStubs::filled(
        &base_order,
        &InstrumentAny::CurrencyPair(instr_a0.clone()),
        None,
        Some(PositionId::new("POS-A")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let fill_a = match fill_a_event {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let pos_a = Position::new(&InstrumentAny::CurrencyPair(instr_a0.clone()), fill_a);

    // Second open position on venue B
    let order_b = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_b0.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from("1"))
        .build();

    let fill_b_event = TestOrderEventStubs::filled(
        &order_b,
        &InstrumentAny::CurrencyPair(instr_b0.clone()),
        None,
        Some(PositionId::new("POS-B")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let fill_b = match fill_b_event {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let pos_b = Position::new(&InstrumentAny::CurrencyPair(instr_b0), fill_b);

    // Closed position on venue A (side Flat + ts_closed)
    let mut pos_closed = pos_a.clone();
    pos_closed.id = PositionId::new("POS-C");
    pos_closed.side = PositionSide::Flat;
    pos_closed.ts_closed = Some(UnixNanos::from(1));

    // Insert into cache
    cache.add_position(&pos_a, OmsType::Netting).unwrap();
    cache.add_position(&pos_b, OmsType::Netting).unwrap();
    cache.add_position(&pos_closed, OmsType::Netting).unwrap();

    // Assertions
    assert_eq!(cache.position_ids(None, None, None, None).len(), 3);

    // Venue filter
    assert_eq!(
        cache.position_ids(Some(&venue_a), None, None, None).len(),
        2
    );

    // Venue + instrument filter
    assert_eq!(
        cache
            .position_ids(Some(&venue_a), Some(&instr_a0.id), None, None)
            .len(),
        2 // open + closed on venue A instrument
    );

    // Open / closed separation
    assert!(
        cache
            .position_open_ids(None, None, None, None)
            .contains(&pos_a.id)
    );
}

// Test order state transitions and cache queries when an order is filled.
//
// This test verifies cache behavior for the complete lifecycle: initialized -> submitted -> accepted -> filled.
// It also tests that position creation and order-position relationships are properly cached.
//
// PRODUCTION CODE BUG: This test likely fails for similar reasons as test_order_when_rejected.
// The cache may incorrectly categorize filled orders or fail to update state properly during
// the order lifecycle transitions.
//
// TODO: Fix cache order state management during order lifecycle. Run this test after fixing
// test_order_when_rejected to see the specific failure.
// The bug is in production code (cache.rs), not in this test.
#[ignore = "Production bug: cache state management during order lifecycle"]
#[rstest]
fn test_order_when_filled(mut cache: Cache, audusd_sim: CurrencyPair) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);
    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = OrderEventAny::Accepted(OrderAccepted::default());
    update_order_with_event(&mut cache, &mut order, accepted);

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        None,
        None,
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, filled);

    let result = cache.order(&order.client_order_id()).unwrap();

    assert!(order.is_closed());
    assert_eq!(&*result, &order);
    drop(result);
    assert_orders_eq(&cache.orders(None, None, None, None, None), &[&order]);
    assert_orders_eq(
        &cache.orders_closed(None, None, None, None, None),
        &[&order],
    );
    assert!(cache.orders_open(None, None, None, None, None).is_empty());
    assert!(
        cache
            .orders_emulated(None, None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .orders_inflight(None, None, None, None, None)
            .is_empty()
    );
    assert!(cache.order_exists(&order.client_order_id()));
    assert!(!cache.is_order_open(&order.client_order_id()));
    assert!(cache.is_order_closed(&order.client_order_id()));
    assert!(!cache.is_order_emulated(&order.client_order_id()));
    assert!(!cache.is_order_inflight(&order.client_order_id()));
    assert!(!cache.is_order_pending_cancel_local(&order.client_order_id()));
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_closed_count(None, None, None, None, None), 1);
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_inflight_count(None, None, None, None, None), 0);
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
    assert_eq!(
        cache.client_order_id(&order.venue_order_id().unwrap()),
        Some(&order.client_order_id())
    );
    assert_eq!(
        cache.venue_order_id(&order.client_order_id()),
        Some(&order.venue_order_id().unwrap())
    );
}

#[rstest]
fn test_get_general_when_empty(cache: Cache) {
    let result = cache.get("A").unwrap();
    assert!(result.is_none());
}

#[rstest]
fn test_add_general_when_value(mut cache: Cache) {
    let key = "A";
    let value = Bytes::from_static(&[0_u8]);
    cache.add(key, value.clone()).unwrap();
    let result = cache.get(key).unwrap();
    assert_eq!(result, Some(&value));
}

#[rstest]
fn test_orders_for_position(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let position_id = PositionId::test_default();
    cache
        .add_order(order.clone(), Some(position_id), None, false)
        .unwrap();
    let result = cache.order(&order.client_order_id()).unwrap();
    assert_eq!(&*result, &order);
    drop(result);
    assert_orders_eq(&cache.orders_for_position(&position_id), &[&order]);
}

#[rstest]
fn test_correct_order_indexing(mut cache: Cache) {
    let binance = Venue::from("BINANCE");
    let bybit = Venue::from("BYBIT");
    let mut orders_generator = TestOrdersGenerator::new(OrderType::Limit);
    orders_generator.add_venue_and_total_instruments(bybit, 10);
    orders_generator.add_venue_and_total_instruments(binance, 10);
    orders_generator.set_orders_per_instrument(2);
    let orders = orders_generator.build();
    // There will be 2 Venues * 10 Instruments * 2 Orders = 40 Orders
    assert_eq!(orders.len(), 40);
    for order in orders {
        cache.add_order(order, None, None, false).unwrap();
    }
    assert_eq!(cache.orders(None, None, None, None, None).len(), 40);
    assert_eq!(cache.orders(Some(&bybit), None, None, None, None).len(), 20);
    assert_eq!(
        cache.orders(Some(&binance), None, None, None, None).len(),
        20
    );
    assert_eq!(
        cache
            .orders(
                Some(&bybit),
                Some(&InstrumentId::from("SYMBOL-0.BYBIT")),
                None,
                None,
                None,
            )
            .len(),
        2
    );
    assert_eq!(
        cache
            .orders(
                Some(&binance),
                Some(&InstrumentId::from("SYMBOL-0.BINANCE")),
                None,
                None,
                None,
            )
            .len(),
        2
    );
}

#[rstest]
fn test_cache_orders_returned_sorted_by_client_order_id(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    // The cache index is AHash-backed for fast lookup, so it iterates in
    // hasher-randomised order. The public Vec returns sort by client_order_id
    // so callers (e.g. own-book replay, cancel-all cascades) see the same
    // sequence across runs.
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    for raw in ["O-303", "O-101", "O-202"] {
        let order = OrderTestBuilder::new(OrderType::Market)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .quantity(Quantity::from(100_000))
            .client_order_id(ClientOrderId::from(raw))
            .build();
        cache.add_order(order, None, None, false).unwrap();
    }

    let returned: Vec<ClientOrderId> = cache
        .orders(None, None, None, None, None)
        .iter()
        .map(|o| o.client_order_id())
        .collect();

    assert_eq!(
        returned,
        vec![
            ClientOrderId::from("O-101"),
            ClientOrderId::from("O-202"),
            ClientOrderId::from("O-303"),
        ],
    );
}

#[rstest]
fn test_cache_positions_returned_sorted_by_position_id(mut cache: Cache, audusd_sim: CurrencyPair) {
    // Mirror of test_cache_orders_returned_sorted_by_client_order_id for the
    // positions path; get_positions_for_ids now sorts by PositionId so the
    // own-book replay and reconciliation flows see a stable sequence.
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    for raw in ["POS-303", "POS-101", "POS-202"] {
        let order = OrderTestBuilder::new(OrderType::Market)
            .instrument_id(instrument.id())
            .side(OrderSide::Buy)
            .quantity(Quantity::from(100_000))
            .build();
        let fill_event = TestOrderEventStubs::filled(
            &order,
            &instrument,
            None,
            Some(PositionId::new(raw)),
            None,
            None,
            None,
            None,
            None,
            None,
        );
        let fill = match fill_event {
            OrderEventAny::Filled(f) => f,
            _ => unreachable!(),
        };
        let position = Position::new(&instrument, fill);
        cache.add_position(&position, OmsType::Hedging).unwrap();
    }

    let returned: Vec<PositionId> = cache
        .positions(None, None, None, None, None)
        .iter()
        .map(|p| p.id)
        .collect();

    assert_eq!(
        returned,
        vec![
            PositionId::new("POS-101"),
            PositionId::new("POS-202"),
            PositionId::new("POS-303"),
        ],
    );
}

#[rstest]
fn test_add_order_with_account_id_populates_account_index() {
    // Verify add_order populates account_orders index when account_id already set
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);
    let account_id = AccountId::new("SIM-001");

    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    // Set account_id before adding (e.g., order loaded from database)
    let submitted = TestOrderEventStubs::submitted(&order, account_id);
    order.apply(submitted).unwrap();

    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    // Verify order is in account_orders index
    assert!(cache.index.account_orders.contains_key(&account_id));
    assert!(
        cache
            .index
            .account_orders
            .get(&account_id)
            .unwrap()
            .contains(&client_order_id)
    );

    // Verify account-filtered query returns the order
    let orders_for_account = cache.orders(None, None, None, Some(&account_id), None);
    assert_eq!(orders_for_account.len(), 1);
    assert!(orders_contains(&orders_for_account, &order));
}

#[rstest]
fn test_add_order_list() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let order_list_id = OrderListId::new("OL-001");
    let order_list = OrderList::new(
        order_list_id,
        instrument.id(),
        order.strategy_id(),
        vec![order.client_order_id()],
        UnixNanos::default(),
    );

    cache.add_order_list(order_list.clone()).unwrap();

    assert!(cache.order_list_exists(&order_list_id));
    assert_eq!(cache.order_list(&order_list_id), Some(&order_list));
    assert!(
        cache
            .order_lists(None, None, None, None)
            .contains(&&order_list)
    );
}

#[rstest]
fn test_add_order_list_when_already_exists_errors() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let order_list_id = OrderListId::new("OL-001");
    let order_list = OrderList::new(
        order_list_id,
        instrument.id(),
        order.strategy_id(),
        vec![order.client_order_id()],
        UnixNanos::default(),
    );

    cache.add_order_list(order_list.clone()).unwrap();
    let result = cache.add_order_list(order_list);

    assert!(result.is_err());
}

#[rstest]
fn test_cache_positions_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_positions()).is_ok());
}

#[rstest]
fn test_position_when_empty(cache: Cache) {
    let position_id = PositionId::from("1");
    let result = cache.position(&position_id);
    assert!(result.is_none());
    assert!(!cache.position_exists(&position_id));
}

#[rstest]
fn test_position_when_some(mut cache: Cache, audusd_sim: CurrencyPair) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-123456")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let position = Position::new(&audusd_sim, filled.into());
    cache.add_position(&position, OmsType::Netting).unwrap();

    let result = cache.position(&position.id).unwrap();
    assert_eq!(result, position);
    assert!(cache.position_exists(&position.id));
    assert_eq!(
        cache.position_id(&order.client_order_id()),
        Some(&position.id)
    );
    let open = cache.positions_open(None, None, None, None, None);
    assert_eq!(open.len(), 1);
    assert_eq!(open[0], position);
    assert!(
        cache
            .positions_closed(None, None, None, None, None)
            .is_empty()
    );
    assert_eq!(cache.positions_open_count(None, None, None, None, None), 1);
    assert_eq!(
        cache.positions_closed_count(None, None, None, None, None),
        0
    );
}

#[rstest]
fn test_position_mut_returns_none_for_missing_position(mut cache: Cache) {
    let position_id = PositionId::from("P-MISSING");
    assert!(cache.position_mut(&position_id).is_none());
}

#[rstest]
fn test_position_owned_returns_none_for_missing_position(cache: Cache) {
    let position_id = PositionId::from("P-MISSING");
    assert!(cache.position_owned(&position_id).is_none());
}

#[rstest]
fn test_position_mut_writes_propagate_to_subsequent_reads(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-MUT-1")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;
    cache.add_position(&position, OmsType::Netting).unwrap();

    cache.position_mut(&position_id).unwrap().quantity = Quantity::from(7);

    assert_eq!(
        cache.position(&position_id).unwrap().quantity,
        Quantity::from(7)
    );
}

#[rstest]
fn test_position_owned_returns_independent_snapshot(mut cache: Cache, audusd_sim: CurrencyPair) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-OWN-1")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;
    let original_quantity = position.quantity;
    cache.add_position(&position, OmsType::Netting).unwrap();

    let snapshot = cache.position_owned(&position_id).unwrap();

    cache.position_mut(&position_id).unwrap().quantity = Quantity::from(7);

    assert_eq!(snapshot.quantity, original_quantity);
    assert_eq!(
        cache.position(&position_id).unwrap().quantity,
        Quantity::from(7)
    );
}

#[rstest]
fn test_update_position_reuses_existing_cell(mut cache: Cache, audusd_sim: CurrencyPair) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-REUSE-1")),
        None,
        None,
        None,
        None,
        None,
        None,
    );
    let position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;
    cache.add_position(&position, OmsType::Netting).unwrap();

    let cell_ptr_before = cache.positions.get(&position_id).unwrap().as_ptr();

    let mut updated = position.clone();
    updated.quantity = Quantity::from(50_000);
    cache.update_position(&updated).unwrap();

    let cell_ptr_after = cache.positions.get(&position_id).unwrap().as_ptr();

    assert_eq!(
        cell_ptr_before, cell_ptr_after,
        "update_position must reuse the existing cell"
    );
    assert_eq!(
        cache.position(&position_id).unwrap().quantity,
        Quantity::from(50_000)
    );
}

// -- DATA ------------------------------------------------------------------------------------

#[rstest]
fn test_cache_currencies_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_currencies()).is_ok());
}

#[rstest]
fn test_cache_instruments_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_instruments()).is_ok());
}

#[rstest]
fn test_instrument_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.instrument(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_instrument_when_some(mut cache: Cache, audusd_sim: CurrencyPair) {
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let result = cache.instrument(&audusd_sim.id);
    assert_eq!(result, Some(&InstrumentAny::CurrencyPair(audusd_sim)));
}

#[rstest]
fn test_instruments_when_empty(cache: Cache) {
    let esz1 = futures_contract_es(None, None);
    let result = cache.instruments(&esz1.id.venue, None);
    assert!(result.is_empty());
}

#[rstest]
fn test_instruments_when_some(mut cache: Cache) {
    let esz1 = futures_contract_es(None, None);
    cache
        .add_instrument(InstrumentAny::FuturesContract(esz1.clone()))
        .unwrap();

    let result1 = cache.instruments(&esz1.id.venue, None);
    let result2 = cache.instruments(&esz1.id.venue, Some(&esz1.underlying));
    assert_eq!(result1, vec![&InstrumentAny::FuturesContract(esz1.clone())]);
    assert_eq!(result2, vec![&InstrumentAny::FuturesContract(esz1.clone())]);
}

fn es_option_contract() -> OptionContract {
    OptionContract::new(
        InstrumentId::from("ESZ1 P4000.GLBX"),
        Symbol::from("ESZ1 P4000"),
        AssetClass::Index,
        Some(Ustr::from("XCME")),
        Ustr::from("ES"),
        OptionKind::Put,
        Price::from("4000.00"),
        Currency::USD(),
        UnixNanos::default(),
        UnixNanos::default(),
        2,
        Price::from("0.01"),
        Quantity::from(1),
        Quantity::from(1),
        None,
        None,
        None,
        None,
        None,
        None,
        None,
        None,
        None::<nautilus_core::Params>,
        UnixNanos::default(),
        UnixNanos::default(),
    )
}

#[rstest]
fn test_instruments_by_parent_filters_by_class(mut cache: Cache) {
    let esz1 = futures_contract_es(None, None);
    let es_put = es_option_contract();
    cache
        .add_instrument(InstrumentAny::FuturesContract(esz1.clone()))
        .unwrap();
    cache
        .add_instrument(InstrumentAny::OptionContract(es_put.clone()))
        .unwrap();

    let futures =
        cache.instruments_by_parent(&esz1.id.venue, &Ustr::from("ES"), InstrumentClass::Future);
    assert_eq!(futures, vec![&InstrumentAny::FuturesContract(esz1.clone())]);

    let options =
        cache.instruments_by_parent(&esz1.id.venue, &Ustr::from("ES"), InstrumentClass::Option);
    assert_eq!(options, vec![&InstrumentAny::OptionContract(es_put)]);
}

#[rstest]
fn test_instruments_by_parent_when_empty(cache: Cache) {
    let result = cache.instruments_by_parent(
        &Venue::from("XCME"),
        &Ustr::from("ES"),
        InstrumentClass::Future,
    );
    assert!(result.is_empty());
}

#[rstest]
fn test_instruments_by_parent_filters_by_root(mut cache: Cache) {
    let esz1 = futures_contract_es(None, None);
    cache
        .add_instrument(InstrumentAny::FuturesContract(esz1.clone()))
        .unwrap();

    let other_root =
        cache.instruments_by_parent(&esz1.id.venue, &Ustr::from("CL"), InstrumentClass::Future);
    assert!(other_root.is_empty());
}

#[rstest]
fn test_cache_synthetics_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_synthetics()).is_ok());
}

#[rstest]
fn test_synthetic_when_empty(cache: Cache) {
    let synth = SyntheticInstrument::default();
    let result = cache.synthetic(&synth.id);
    assert!(result.is_none());
}

#[rstest]
fn test_synthetic_when_some(mut cache: Cache) {
    let synth = SyntheticInstrument::default();
    cache.add_synthetic(synth.clone()).unwrap();
    let result = cache.synthetic(&synth.id);
    assert_eq!(result, Some(&synth));
}

#[rstest]
fn test_order_book_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.order_book(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_order_book_when_some(mut cache: Cache, audusd_sim: CurrencyPair) {
    let book = OrderBook::new(audusd_sim.id, BookType::L2_MBP);
    cache.add_order_book(book.clone()).unwrap();
    let result = cache.order_book(&audusd_sim.id);
    assert_eq!(result, Some(&book));
}

#[rstest]
fn test_order_book_mut_when_empty(mut cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.order_book_mut(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_order_book_mut_when_some(mut cache: Cache, audusd_sim: CurrencyPair) {
    let mut book = OrderBook::new(audusd_sim.id, BookType::L2_MBP);
    cache.add_order_book(book.clone()).unwrap();
    let result = cache.order_book_mut(&audusd_sim.id);
    assert_eq!(result, Some(&mut book));
}

#[cfg(feature = "defi")]
#[fixture]
fn test_pool() -> Pool {
    let chain = Arc::new(chains::ETHEREUM.clone());
    let dex = Dex::new(
        chains::ETHEREUM.clone(),
        DexType::UniswapV3,
        "0x1F98431c8aD98523631AE4a59f267346ea31F984",
        0,
        AmmType::CLAMM,
        "PoolCreated(address,address,uint24,int24,address)",
        "Swap(address,address,int256,int256,uint160,uint128,int24)",
        "Mint(address,address,int24,int24,uint128,uint256,uint256)",
        "Burn(address,int24,int24,uint128,uint256,uint256)",
        "Collect(address,address,int24,int24,uint128,uint128)",
    );

    let token0 = Token::new(
        chain.clone(),
        "0xA0b86a33E6441b936662bb6B5d1F8Fb0E2b57A5D"
            .parse()
            .unwrap(),
        "Wrapped Ether".to_string(),
        "WETH".to_string(),
        18,
    );

    let token1 = Token::new(
        chain.clone(),
        "0xdAC17F958D2ee523a2206206994597C13D831ec7"
            .parse()
            .unwrap(),
        "Tether USD".to_string(),
        "USDT".to_string(),
        6,
    );

    let pool_address = "0x11b815efB8f581194ae79006d24E0d814B7697F6"
        .parse()
        .unwrap();
    let pool_identifier: PoolIdentifier = "0x11b815efB8f581194ae79006d24E0d814B7697F6"
        .parse()
        .unwrap();
    Pool::new(
        chain,
        Arc::new(dex),
        pool_address,
        pool_identifier,
        12345678,
        token0,
        token1,
        Some(3000),
        Some(60),
        UnixNanos::from(1_234_567_890_000_000_000u64),
    )
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_when_empty(cache: Cache, test_pool: Pool) {
    let instrument_id = test_pool.instrument_id;
    let result = cache.pool(&instrument_id);
    assert!(result.is_none());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_when_some(mut cache: Cache, test_pool: Pool) {
    let instrument_id = test_pool.instrument_id;
    cache.add_pool(test_pool.clone()).unwrap();
    let result = cache.pool(&instrument_id);
    assert_eq!(result, Some(&test_pool));
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_mut_when_empty(mut cache: Cache, test_pool: Pool) {
    let instrument_id = test_pool.instrument_id;
    let result = cache.pool_mut(&instrument_id);
    assert!(result.is_none());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_mut_when_some(mut cache: Cache, test_pool: Pool) {
    let instrument_id = test_pool.instrument_id;
    cache.add_pool(test_pool).unwrap();
    let result = cache.pool_mut(&instrument_id);

    assert!(result.is_some());
    if let Some(pool_ref) = result {
        assert_eq!(pool_ref.fee.unwrap(), 3000);
    }
}

#[cfg(feature = "defi")]
#[rstest]
fn test_add_pool(mut cache: Cache, test_pool: Pool) {
    let instrument_id = test_pool.instrument_id;

    cache.add_pool(test_pool.clone()).unwrap();

    let cached_pool = cache.pool(&instrument_id);
    assert!(cached_pool.is_some());
    assert_eq!(cached_pool.unwrap(), &test_pool);
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_ids_when_empty(cache: Cache, test_pool: Pool) {
    let result = cache.pool_ids(Some(&test_pool.instrument_id.venue));
    assert!(result.is_empty());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_ids_when_some(mut cache: Cache, test_pool: Pool) {
    let venue = test_pool.instrument_id.venue;
    cache.add_pool(test_pool.clone()).unwrap();

    let result1 = cache.pool_ids(None);
    let result2 = cache.pool_ids(Some(&venue));
    assert_eq!(result1, vec![test_pool.instrument_id]);
    assert_eq!(result2, vec![test_pool.instrument_id]);
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pools_when_empty(cache: Cache, test_pool: Pool) {
    let result = cache.pools(Some(&test_pool.instrument_id.venue));
    assert!(result.is_empty());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pools_when_some(mut cache: Cache, test_pool: Pool) {
    let venue = test_pool.instrument_id.venue;
    cache.add_pool(test_pool.clone()).unwrap();

    let result1 = cache.pools(None);
    let result2 = cache.pools(Some(&venue));
    assert_eq!(result1, vec![&test_pool]);
    assert_eq!(result2, vec![&test_pool]);
}

#[cfg(feature = "defi")]
#[fixture]
fn test_pool_profiler(test_pool: Pool) -> PoolProfiler {
    PoolProfiler::new(Arc::new(test_pool))
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_when_empty(cache: Cache, test_pool_profiler: PoolProfiler) {
    let instrument_id = test_pool_profiler.pool.instrument_id;
    let result = cache.pool_profiler(&instrument_id);
    assert!(result.is_none());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_when_some(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let instrument_id = test_pool_profiler.pool.instrument_id;
    cache.add_pool_profiler(test_pool_profiler).unwrap();
    let result = cache.pool_profiler(&instrument_id);
    assert!(result.is_some());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_mut_when_empty(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let instrument_id = test_pool_profiler.pool.instrument_id;
    let result = cache.pool_profiler_mut(&instrument_id);
    assert!(result.is_none());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_mut_when_some(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let instrument_id = test_pool_profiler.pool.instrument_id;
    cache.add_pool_profiler(test_pool_profiler).unwrap();
    let result = cache.pool_profiler_mut(&instrument_id);
    assert!(result.is_some());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_add_pool_profiler(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let instrument_id = test_pool_profiler.pool.instrument_id;

    cache.add_pool_profiler(test_pool_profiler).unwrap();

    let cached_profiler = cache.pool_profiler(&instrument_id);
    assert!(cached_profiler.is_some());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_ids_when_empty(cache: Cache, test_pool_profiler: PoolProfiler) {
    let result = cache.pool_profiler_ids(Some(&test_pool_profiler.pool.instrument_id.venue));
    assert!(result.is_empty());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profiler_ids_when_some(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let venue = test_pool_profiler.pool.instrument_id.venue;
    cache.add_pool_profiler(test_pool_profiler.clone()).unwrap();

    let result1 = cache.pool_profiler_ids(None);
    let result2 = cache.pool_profiler_ids(Some(&venue));
    assert_eq!(result1, vec![test_pool_profiler.pool.instrument_id]);
    assert_eq!(result2, vec![test_pool_profiler.pool.instrument_id]);
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profilers_when_empty(cache: Cache, test_pool_profiler: PoolProfiler) {
    let result = cache.pool_profilers(Some(&test_pool_profiler.pool.instrument_id.venue));
    assert!(result.is_empty());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_pool_profilers_when_some(mut cache: Cache, test_pool_profiler: PoolProfiler) {
    let venue = test_pool_profiler.pool.instrument_id.venue;
    cache.add_pool_profiler(test_pool_profiler).unwrap();

    let result1 = cache.pool_profilers(None);
    let result2 = cache.pool_profilers(Some(&venue));
    assert_eq!(result1.len(), 1);
    assert_eq!(result2.len(), 1);
}

#[rstest]
#[case(PriceType::Bid)]
#[case(PriceType::Ask)]
#[case(PriceType::Mid)]
#[case(PriceType::Last)]
#[case(PriceType::Mark)]
fn test_price_when_empty(cache: Cache, audusd_sim: CurrencyPair, #[case] price_type: PriceType) {
    let result = cache.price(&audusd_sim.id, price_type);
    assert!(result.is_none());
}

#[rstest]
fn test_price_when_some(mut cache: Cache, audusd_sim: CurrencyPair) {
    let mark_price = MarkPriceUpdate::new(
        audusd_sim.id,
        Price::from("1.00000"),
        UnixNanos::from(5),
        UnixNanos::from(10),
    );
    cache.add_mark_price(mark_price).unwrap();
    let result = cache.price(&audusd_sim.id, PriceType::Mark);
    assert_eq!(result, Some(mark_price.value));
}

#[rstest]
fn test_quote_tick_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.quote(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_quote_tick_when_some(mut cache: Cache) {
    let quote = QuoteTick::default();
    cache.add_quote(quote).unwrap();
    let result = cache.quote(&quote.instrument_id);
    assert_eq!(result, Some(&quote));
}

#[rstest]
fn test_quote_ticks_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.quotes(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_quote_ticks_when_some(mut cache: Cache) {
    let quotes = vec![
        QuoteTick::default(),
        QuoteTick::default(),
        QuoteTick::default(),
    ];
    cache.add_quotes(&quotes).unwrap();
    let result = cache.quotes(&quotes[0].instrument_id);
    assert_eq!(result, Some(quotes));
}

#[rstest]
fn test_trade_tick_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.trade(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_trade_tick_when_some(mut cache: Cache) {
    let trade = TradeTick::default();
    cache.add_trade(trade).unwrap();
    let result = cache.trade(&trade.instrument_id);
    assert_eq!(result, Some(&trade));
}

#[rstest]
fn test_trade_ticks_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.trades(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_trade_ticks_when_some(mut cache: Cache) {
    let trades = vec![
        TradeTick::default(),
        TradeTick::default(),
        TradeTick::default(),
    ];
    cache.add_trades(&trades).unwrap();
    let result = cache.trades(&trades[0].instrument_id);
    assert_eq!(result, Some(trades));
}

#[rstest]
fn test_mark_price_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.mark_price(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_mark_prices_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.mark_prices(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_index_price_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.index_price(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_index_prices_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.index_prices(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_funding_rate_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    let result = cache.funding_rate(&audusd_sim.id);
    assert!(result.is_none());
}

#[rstest]
fn test_add_funding_rate(mut cache: Cache, audusd_sim: CurrencyPair) {
    let funding_rate = FundingRateUpdate::new(
        audusd_sim.id,
        "0.0001".parse().unwrap(),
        None,
        None,
        UnixNanos::from(5),
        UnixNanos::from(10),
    );

    cache.add_funding_rate(funding_rate).unwrap();

    let result = cache.funding_rate(&audusd_sim.id);
    assert_eq!(result, Some(&funding_rate));
}

#[rstest]
fn test_add_funding_rate_updates_existing(mut cache: Cache, audusd_sim: CurrencyPair) {
    let funding_rate1 = FundingRateUpdate::new(
        audusd_sim.id,
        "0.0001".parse().unwrap(),
        None,
        None,
        UnixNanos::from(5),
        UnixNanos::from(10),
    );

    let funding_rate2 = FundingRateUpdate::new(
        audusd_sim.id,
        "0.0002".parse().unwrap(),
        None,
        None,
        UnixNanos::from(15),
        UnixNanos::from(20),
    );

    cache.add_funding_rate(funding_rate1).unwrap();
    cache.add_funding_rate(funding_rate2).unwrap();

    let result = cache.funding_rate(&audusd_sim.id);
    assert_eq!(result, Some(&funding_rate2));
}

#[rstest]
fn test_instrument_status_when_empty(cache: Cache, audusd_sim: CurrencyPair) {
    assert!(cache.instrument_status(&audusd_sim.id).is_none());
    assert!(cache.instrument_statuses(&audusd_sim.id).is_none());
}

#[rstest]
fn test_add_instrument_status(mut cache: Cache, audusd_sim: CurrencyPair) {
    let status = InstrumentStatus::new(
        audusd_sim.id,
        MarketStatusAction::Trading,
        UnixNanos::from(5),
        UnixNanos::from(10),
        None,
        None,
        Some(true),
        Some(true),
        None,
    );

    cache.add_instrument_status(status).unwrap();

    assert_eq!(cache.instrument_status(&audusd_sim.id), Some(&status));
    assert_eq!(
        cache.instrument_statuses(&audusd_sim.id),
        Some(vec![status])
    );
}

#[rstest]
fn test_add_instrument_status_keeps_time_series(mut cache: Cache, audusd_sim: CurrencyPair) {
    let status1 = InstrumentStatus::new(
        audusd_sim.id,
        MarketStatusAction::PreOpen,
        UnixNanos::from(5),
        UnixNanos::from(10),
        None,
        None,
        Some(false),
        Some(false),
        None,
    );
    let status2 = InstrumentStatus::new(
        audusd_sim.id,
        MarketStatusAction::Trading,
        UnixNanos::from(15),
        UnixNanos::from(20),
        None,
        None,
        Some(true),
        Some(true),
        None,
    );

    cache.add_instrument_status(status1).unwrap();
    cache.add_instrument_status(status2).unwrap();

    // Latest status first (push_front semantics)
    assert_eq!(cache.instrument_status(&audusd_sim.id), Some(&status2));
    assert_eq!(
        cache.instrument_statuses(&audusd_sim.id),
        Some(vec![status2, status1]),
    );
}

#[rstest]
fn test_bar_when_empty(cache: Cache) {
    let bar = Bar::default();
    let result = cache.bar(&bar.bar_type);
    assert!(result.is_none());
}

#[rstest]
fn test_bar_when_some(mut cache: Cache) {
    let bar = Bar::default();
    cache.add_bar(bar).unwrap();
    let result = cache.bar(&bar.bar_type);
    assert_eq!(result, Some(&bar));
}

#[rstest]
fn test_bars_when_empty(cache: Cache) {
    let bar = Bar::default();
    let result = cache.bars(&bar.bar_type);
    assert!(result.is_none());
}

#[rstest]
fn test_bars_when_some(mut cache: Cache) {
    let bars = vec![Bar::default(), Bar::default(), Bar::default()];
    cache.add_bars(&bars).unwrap();
    let result = cache.bars(&bars[0].bar_type);
    assert_eq!(result, Some(bars));
}

// -- ACCOUNT ---------------------------------------------------------------------------------

#[rstest]
fn test_cache_accounts_when_no_database(mut cache: Cache) {
    assert!(futures::executor::block_on(cache.cache_accounts()).is_ok());
}

#[rstest]
fn test_cache_add_account(mut cache: Cache) {
    let account = AccountAny::default();
    cache.add_account(account.clone()).unwrap();
    let result = cache.account(&account.id());
    assert!(result.is_some());
    assert_eq!(*result.unwrap(), account);
}

#[rstest]
fn test_cache_accounts_when_no_accounts_returns_empty(cache: Cache) {
    let result = cache.accounts(&AccountId::test_default());
    assert!(result.is_empty());
}

#[rstest]
fn test_cache_account_for_venue_returns_empty(cache: Cache) {
    let venue = Venue::test_default();
    let result = cache.account_for_venue(&venue);
    assert!(result.is_none());
}

#[rstest]
fn test_cache_account_for_venue_return_correct(mut cache: Cache) {
    let account = AccountAny::default();
    let venue = account.last_event().unwrap().account_id.get_issuer();
    cache.add_account(account.clone()).unwrap();
    let result = cache.account_for_venue(&venue);
    assert!(result.is_some());
    assert_eq!(*result.unwrap(), account);
}

#[rstest]
fn test_cache_take_account_returns_none_for_unknown(mut cache: Cache) {
    let result = cache.take_account(&AccountId::test_default());
    assert!(result.is_none());
}

#[rstest]
fn test_cache_update_account_owned_restores_venue_index(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let venue = account_id.get_issuer();

    cache.add_account(account).unwrap();
    let account = cache.take_account(&account_id).unwrap();

    assert!(cache.account(&account_id).is_none());
    assert!(cache.account_for_venue(&venue).is_none());

    cache.update_account_owned(account.clone()).unwrap();

    assert_eq!(cache.account(&account_id).unwrap(), account);
    assert_eq!(cache.account_for_venue(&venue).unwrap(), account);
}

#[rstest]
fn test_cache_update_account_state_adds_new_account(mut cache: Cache) {
    let account = AccountAny::default();
    let event = account.last_event().unwrap();
    let account_id = event.account_id;
    let venue = account_id.get_issuer();

    cache.update_account_state(&event).unwrap();

    let cached = cache.account(&account_id).unwrap();
    assert_eq!(cached.id(), account_id);
    assert_eq!(cached.events(), vec![event]);
    assert_eq!(cached.balances(), account.balances());
    assert_eq!(cache.account_for_venue(&venue).unwrap().id(), account_id);
}

#[rstest]
fn test_cache_update_account_state_apply_failure_leaves_account_intact(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let starting_balances = account.balances();
    let event = AccountState::new(
        account_id,
        AccountType::Cash,
        vec![AccountBalance::new(
            Money::from("-1 USD"),
            Money::from("0 USD"),
            Money::from("-1 USD"),
        )],
        vec![],
        true,
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        Some(Currency::USD()),
    );

    cache.add_account(account).unwrap();
    let result = cache.update_account_state(&event);
    let cached = cache.account(&account_id).unwrap();

    let error = result.unwrap_err().to_string();
    assert!(error.contains("balance would be negative"));
    assert!(error.contains("borrowing not allowed"));
    assert_eq!(cached.balances(), starting_balances);
    assert_eq!(cached.events().len(), 1);
}

fn make_cash_account_state(account_id: AccountId, total_usd: &str) -> AccountState {
    AccountState::new(
        account_id,
        AccountType::Cash,
        vec![AccountBalance::new(
            Money::from(format!("{total_usd} USD")),
            Money::from("0 USD"),
            Money::from(format!("{total_usd} USD")),
        )],
        vec![],
        true,
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        Some(Currency::USD()),
    )
}

#[rstest]
fn test_cache_account_mut_returns_none_for_missing_account(mut cache: Cache) {
    assert!(cache.account_mut(&AccountId::test_default()).is_none());
}

#[rstest]
fn test_cache_account_owned_returns_none_for_missing_account(cache: Cache) {
    assert!(cache.account_owned(&AccountId::test_default()).is_none());
}

#[rstest]
fn test_cache_account_for_venue_owned_returns_none_for_missing_venue(cache: Cache) {
    assert!(
        cache
            .account_for_venue_owned(&Venue::test_default())
            .is_none()
    );
}

#[rstest]
fn test_cache_account_mut_writes_propagate_to_subsequent_reads(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let starting_event_count = account.events().len();
    cache.add_account(account).unwrap();

    let event = make_cash_account_state(account_id, "2000");

    cache
        .account_mut(&account_id)
        .unwrap()
        .apply(event.clone())
        .unwrap();

    let cached = cache.account(&account_id).unwrap();
    assert_eq!(cached.events().len(), starting_event_count + 1);
    assert_eq!(cached.events().last().unwrap(), &event);
}

#[rstest]
fn test_cache_account_owned_returns_independent_snapshot(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let starting_event_count = account.events().len();
    cache.add_account(account).unwrap();

    let snapshot = cache.account_owned(&account_id).unwrap();

    let event = make_cash_account_state(account_id, "2000");
    cache
        .account_mut(&account_id)
        .unwrap()
        .apply(event)
        .unwrap();

    assert_eq!(snapshot.events().len(), starting_event_count);
    assert_eq!(
        cache.account(&account_id).unwrap().events().len(),
        starting_event_count + 1
    );
}

#[rstest]
fn test_cache_account_for_venue_owned_returns_independent_snapshot(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let venue = account_id.get_issuer();
    let starting_event_count = account.events().len();
    cache.add_account(account).unwrap();

    let snapshot = cache.account_for_venue_owned(&venue).unwrap();

    let event = make_cash_account_state(account_id, "2000");
    cache
        .account_mut(&account_id)
        .unwrap()
        .apply(event)
        .unwrap();

    assert_eq!(snapshot.events().len(), starting_event_count);
    assert_eq!(
        cache.account_for_venue(&venue).unwrap().events().len(),
        starting_event_count + 1
    );
}

#[rstest]
fn test_update_account_reuses_existing_cell(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    cache.add_account(account.clone()).unwrap();

    let cell_ptr_before = cache.accounts.get(&account_id).unwrap().as_ptr();

    cache.update_account(&account).unwrap();

    let cell_ptr_after = cache.accounts.get(&account_id).unwrap().as_ptr();

    assert_eq!(
        cell_ptr_before, cell_ptr_after,
        "update_account must reuse the existing cell"
    );
}

#[rstest]
fn test_update_account_state_grows_event_log_in_place(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    let starting_event_count = account.events().len();
    cache.add_account(account).unwrap();

    let cell_ptr_before = cache.accounts.get(&account_id).unwrap().as_ptr();

    let event = make_cash_account_state(account_id, "2000");
    cache.update_account_state(&event).unwrap();

    let cell_ptr_after = cache.accounts.get(&account_id).unwrap().as_ptr();

    let cached = cache.account(&account_id).unwrap();
    assert_eq!(cached.events().len(), starting_event_count + 1);
    assert_eq!(cached.events().last().unwrap(), &event);
    assert_eq!(
        cell_ptr_before, cell_ptr_after,
        "update_account_state must apply in place, not replace the cell"
    );
}

#[rstest]
#[should_panic(expected = "sole owner")]
fn test_take_account_panics_when_cell_aliased(mut cache: Cache) {
    let account = AccountAny::default();
    let account_id = account.id();
    cache.add_account(account).unwrap();

    // Manufacture an aliased SharedCell handle by cloning the inner Rc.
    // This violates the sole-owner invariant; take_account must panic.
    let _alias = cache.accounts.get(&account_id).unwrap().clone();
    let _ = cache.take_account(&account_id);
}

#[rstest]
#[case::matching(true)]
#[case::non_matching(false)]
fn test_cache_accounts_filters_by_id(mut cache: Cache, #[case] matching: bool) {
    let account = AccountAny::default();
    let account_id = account.id();
    cache.add_account(account.clone()).unwrap();

    if matching {
        let result = cache.accounts(&account_id);
        assert_eq!(result.len(), 1);
        assert_eq!(*result[0], account);
    } else {
        let result = cache.accounts(&AccountId::from("OTHER-001"));
        assert!(result.is_empty());
    }
}

#[rstest]
fn test_get_mark_xrate_returns_none(cache: Cache) {
    // When no mark xrate is set for (USD, EUR), it should return None
    assert!(
        cache
            .get_mark_xrate(Currency::USD(), Currency::EUR())
            .is_none()
    );
}

#[rstest]
fn test_set_and_get_mark_xrate(mut cache: Cache) {
    // Set a mark xrate for (USD, EUR) and check both forward and inverse rates
    let xrate = 1.25;
    cache.set_mark_xrate(Currency::USD(), Currency::EUR(), xrate);
    assert_eq!(
        cache.get_mark_xrate(Currency::USD(), Currency::EUR()),
        Some(xrate)
    );
    assert_eq!(
        cache.get_mark_xrate(Currency::EUR(), Currency::USD()),
        Some(1.0 / xrate)
    );
}

#[rstest]
fn test_clear_mark_xrate(mut cache: Cache) {
    // Set a rate and then clear the forward key
    let xrate = 1.25;
    cache.set_mark_xrate(Currency::USD(), Currency::EUR(), xrate);
    assert!(
        cache
            .get_mark_xrate(Currency::USD(), Currency::EUR())
            .is_some()
    );
    cache.clear_mark_xrate(Currency::USD(), Currency::EUR());
    assert!(
        cache
            .get_mark_xrate(Currency::USD(), Currency::EUR())
            .is_none()
    );
    assert_eq!(
        cache.get_mark_xrate(Currency::EUR(), Currency::USD()),
        Some(1.0 / xrate)
    );
}

#[rstest]
fn test_clear_mark_xrates(mut cache: Cache) {
    // Set two mark xrates and then clear them all
    cache.set_mark_xrate(Currency::USD(), Currency::EUR(), 1.25);
    cache.set_mark_xrate(Currency::AUD(), Currency::USD(), 0.75);
    cache.clear_mark_xrates();
    assert!(
        cache
            .get_mark_xrate(Currency::USD(), Currency::EUR())
            .is_none()
    );
    assert!(
        cache
            .get_mark_xrate(Currency::EUR(), Currency::USD())
            .is_none()
    );
    assert!(
        cache
            .get_mark_xrate(Currency::AUD(), Currency::USD())
            .is_none()
    );
    assert!(
        cache
            .get_mark_xrate(Currency::USD(), Currency::AUD())
            .is_none()
    );
}

#[rstest]
#[should_panic(expected = "xrate was zero")]
fn test_set_mark_xrate_panics_on_zero(mut cache: Cache) {
    // Setting a mark xrate of zero should panic
    cache.set_mark_xrate(Currency::USD(), Currency::EUR(), 0.0);
}

#[rstest]
fn test_purge_order() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create an order and fill to generate a position
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        Some(TradeId::new("T-1")),
        Some(PositionId::new("P-123456")),
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );

    cache.add_order(order, None, None, false).unwrap();

    let mut position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;
    cache.add_position(&position, OmsType::Netting).unwrap();

    // Close the position to test purging from closed positions
    let order_close = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-19700101-000000-001-001-2"))
        .build();

    let filled_close = TestOrderEventStubs::filled(
        &order_close,
        &audusd_sim,
        Some(TradeId::new("T-2")),
        Some(position_id),
        Some(Price::from("1.00010")),
        None,
        None,
        None,
        None,
        None,
    );

    position.apply(&filled_close.into());
    cache.update_position(&position).unwrap();

    // Verify position is now closed
    assert!(position.is_closed());

    // Verify the order exists
    assert!(cache.order_exists(&client_order_id));
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);

    // Add the closing order to cache so it can be purged
    let client_order_id_close = order_close.client_order_id();
    cache
        .add_order(order_close, Some(position_id), None, false)
        .unwrap();

    // Purge both orders - fills should NOT be purged from the position
    cache.purge_order(client_order_id);
    cache.purge_order(client_order_id_close);

    // Verify the orders are gone
    assert!(!cache.order_exists(&client_order_id));
    assert!(!cache.order_exists(&client_order_id_close));
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 0);
    // Verify position fills are preserved (purge_order doesn't touch position fills)
    assert_eq!(cache.position(&position_id).unwrap().event_count(), 2);
}

#[rstest]
fn test_purge_open_order_skips_purge() {
    // Test that attempting to purge an open order is prevented by the guard
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create and accept an order to make it open
    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();
    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = OrderEventAny::Accepted(OrderAccepted::default());
    update_order_with_event(&mut cache, &mut order, accepted);

    // Verify order is open
    assert!(order.is_open());
    assert!(cache.order_exists(&client_order_id));
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);

    // Attempt to purge the open order - should be prevented by guard
    cache.purge_order(client_order_id);

    // Verify the order still exists (guard prevented purge)
    assert!(cache.order_exists(&client_order_id));
    assert_eq!(cache.orders_total_count(None, None, None, None, None), 1);
    assert!(cache.order_exists(&client_order_id));
}

#[rstest]
fn test_purge_position() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create an order and fill to generate a position
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-123456")),
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );

    let mut position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;

    // Add position to cache
    cache.add_position(&position, OmsType::Netting).unwrap();

    // Verify the position exists and is open
    assert!(cache.position_exists(&position_id));
    assert!(position.is_open());
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 1);

    // Close the position first (create a closing order and fill)
    let order_close = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-19700101-000000-001-001-2"))
        .build();

    let filled_close = TestOrderEventStubs::filled(
        &order_close,
        &audusd_sim,
        Some(TradeId::new("T-2")),
        Some(position_id),
        Some(Price::from("1.00010")),
        None,
        None,
        None,
        None,
        None,
    );

    position.apply(&filled_close.into());
    cache.update_position(&position).unwrap();

    // Verify position is now closed
    assert!(position.is_closed());

    // Purge the position
    cache.purge_position(position_id);

    // Verify the position is gone
    assert!(!cache.position_exists(&position_id));
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 0);
}

#[rstest]
fn test_purge_open_position_skips_purge() {
    // Test that attempting to purge an open position is prevented by the guard
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create an order and fill to generate an open position
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        Some(PositionId::new("P-123456")),
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );

    let position = Position::new(&audusd_sim, filled.into());
    let position_id = position.id;

    cache.add_position(&position, OmsType::Netting).unwrap();

    // Verify position is open
    assert!(position.is_open());
    assert!(cache.position_exists(&position_id));
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 1);
    assert_eq!(position.event_count(), 1);

    // Attempt to purge the open position - should be prevented by guard
    cache.purge_position(position_id);

    // Verify the position still exists (guard prevented purge)
    assert!(cache.position_exists(&position_id));
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 1);
    assert!(cache.position(&position_id).is_some());
    // Verify events are preserved
    assert_eq!(cache.position(&position_id).unwrap().event_count(), 1);
}

#[rstest]
fn test_purge_instrument_removes_from_cache_and_indices() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument_id = audusd_sim.id;
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    cache.add_instrument(instrument.clone()).unwrap();

    // Populate every cache-owned per-instrument map so we can confirm each one is
    // cleaned up by the purge.
    let quote = QuoteTick {
        instrument_id,
        ..QuoteTick::default()
    };
    cache.add_quote(quote).unwrap();

    let trade = TradeTick {
        instrument_id,
        ..TradeTick::default()
    };
    cache.add_trade(trade).unwrap();

    let bar_type = BarType::from(format!("{instrument_id}-1-MINUTE-LAST-EXTERNAL").as_str());
    let bar = Bar {
        bar_type,
        ..Bar::default()
    };
    cache.add_bar(bar).unwrap();

    let order_book = OrderBook::new(instrument_id, BookType::L2_MBP);
    cache.add_order_book(order_book).unwrap();

    let mark_price = MarkPriceUpdate::new(
        instrument_id,
        Price::from("1.00000"),
        UnixNanos::from(5),
        UnixNanos::from(10),
    );
    cache.add_mark_price(mark_price).unwrap();

    let index_price = IndexPriceUpdate::new(
        instrument_id,
        Price::from("1.00000"),
        UnixNanos::from(5),
        UnixNanos::from(10),
    );
    cache.add_index_price(index_price).unwrap();

    let funding_rate = FundingRateUpdate::new(
        instrument_id,
        "0.0001".parse().unwrap(),
        None,
        None,
        UnixNanos::from(5),
        UnixNanos::from(10),
    );
    cache.add_funding_rate(funding_rate).unwrap();

    let status = InstrumentStatus::new(
        instrument_id,
        MarketStatusAction::Trading,
        UnixNanos::from(5),
        UnixNanos::from(10),
        None,
        None,
        Some(true),
        Some(true),
        None,
    );
    cache.add_instrument_status(status).unwrap();

    // Add a fully filled order and its filled-then-closed position so every associated
    // order is in `orders_closed` and every position is in `positions_closed`.
    let mut order_open = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument_id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let position_id = PositionId::new("P-PURGE-1");
    let account_id = AccountId::new("SIM-001");
    cache
        .add_order(order_open.clone(), None, None, false)
        .unwrap();
    let submitted_open = TestOrderEventStubs::submitted(&order_open, account_id);
    update_order_with_event(&mut cache, &mut order_open, submitted_open);
    let accepted_open =
        TestOrderEventStubs::accepted(&order_open, account_id, VenueOrderId::new("V-1"));
    update_order_with_event(&mut cache, &mut order_open, accepted_open);
    let filled_open = TestOrderEventStubs::filled(
        &order_open,
        &instrument,
        Some(TradeId::new("T-1")),
        Some(position_id),
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order_open, filled_open.clone());
    assert!(order_open.is_closed());

    let mut position = Position::new(&instrument, filled_open.into());
    cache.add_position(&position, OmsType::Netting).unwrap();

    let mut order_close = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument_id)
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-19700101-000000-001-001-2"))
        .build();
    cache
        .add_order(order_close.clone(), Some(position_id), None, false)
        .unwrap();
    let submitted_close = TestOrderEventStubs::submitted(&order_close, account_id);
    update_order_with_event(&mut cache, &mut order_close, submitted_close);
    let accepted_close =
        TestOrderEventStubs::accepted(&order_close, account_id, VenueOrderId::new("V-2"));
    update_order_with_event(&mut cache, &mut order_close, accepted_close);
    let filled_close = TestOrderEventStubs::filled(
        &order_close,
        &instrument,
        Some(TradeId::new("T-2")),
        Some(position_id),
        Some(Price::from("1.00010")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order_close, filled_close.clone());
    assert!(order_close.is_closed());

    position.apply(&filled_close.into());
    cache.update_position(&position).unwrap();
    assert!(position.is_closed());

    assert!(cache.instrument(&instrument_id).is_some());
    assert!(cache.has_quote_ticks(&instrument_id));
    assert!(cache.has_trade_ticks(&instrument_id));
    assert!(cache.has_bars(&bar_type));
    assert!(cache.order_book(&instrument_id).is_some());
    assert!(cache.mark_price(&instrument_id).is_some());
    assert!(cache.index_price(&instrument_id).is_some());
    assert!(cache.funding_rate(&instrument_id).is_some());
    assert!(cache.instrument_status(&instrument_id).is_some());

    cache.purge_instrument(instrument_id);

    assert!(cache.instrument(&instrument_id).is_none());
    assert!(!cache.has_quote_ticks(&instrument_id));
    assert!(!cache.has_trade_ticks(&instrument_id));
    assert!(!cache.has_bars(&bar_type));
    assert!(cache.order_book(&instrument_id).is_none());
    assert!(cache.mark_price(&instrument_id).is_none());
    assert!(cache.index_price(&instrument_id).is_none());
    assert!(cache.funding_rate(&instrument_id).is_none());
    assert!(cache.instrument_status(&instrument_id).is_none());
    assert!(!cache.index.instrument_orders.contains_key(&instrument_id));
    assert!(
        !cache
            .index
            .instrument_positions
            .contains_key(&instrument_id)
    );
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_instrument_when_not_in_cache_is_noop() {
    let mut cache = Cache::default();
    let instrument_id = InstrumentId::from("AUD/USD.SIM");

    cache.purge_instrument(instrument_id);

    assert!(cache.instrument(&instrument_id).is_none());
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_instrument_refuses_when_orders_open() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument_id = audusd_sim.id;
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    cache.add_instrument(instrument).unwrap();

    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument_id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    cache.add_order(order.clone(), None, None, false).unwrap();

    update_order_with_event(
        &mut cache,
        &mut order,
        OrderEventAny::Submitted(OrderSubmitted::default()),
    );
    update_order_with_event(
        &mut cache,
        &mut order,
        OrderEventAny::Accepted(OrderAccepted::default()),
    );
    assert!(order.is_open());

    cache.purge_instrument(instrument_id);

    assert!(cache.instrument(&instrument_id).is_some());
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_instrument_refuses_when_orders_initialized_but_not_open() {
    // Regression: orders in non-terminal states like INITIALIZED/SUBMITTED are not in
    // `orders_open`, but purging would still leave them dangling without an instrument.
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument_id = audusd_sim.id;
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    cache.add_instrument(instrument).unwrap();

    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument_id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();
    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(!order.is_open());
    assert!(!order.is_closed());

    cache.purge_instrument(instrument_id);

    assert!(cache.instrument(&instrument_id).is_some());
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_instrument_refuses_when_positions_open() {
    // Take the order through to FILLED so the order guard passes; the position remains
    // open and must be the reason the purge is refused.
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument_id = audusd_sim.id;
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    cache.add_instrument(instrument.clone()).unwrap();

    let position_id = PositionId::new("P-OPEN-1");
    let account_id = AccountId::new("SIM-001");
    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument_id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    cache
        .add_order(order.clone(), Some(position_id), None, false)
        .unwrap();
    let submitted = TestOrderEventStubs::submitted(&order, account_id);
    update_order_with_event(&mut cache, &mut order, submitted);
    let accepted = TestOrderEventStubs::accepted(&order, account_id, VenueOrderId::new("V-1"));
    update_order_with_event(&mut cache, &mut order, accepted);
    let filled = TestOrderEventStubs::filled(
        &order,
        &instrument,
        None,
        Some(position_id),
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, filled.clone());
    assert!(order.is_closed());

    let position = Position::new(&instrument, filled.into());
    cache.add_position(&position, OmsType::Netting).unwrap();
    assert!(position.is_open());

    cache.purge_instrument(instrument_id);

    assert!(cache.instrument(&instrument_id).is_some());
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_instrument_clears_synthetic_only() {
    // Synthetic instruments share the InstrumentId space; the absence guard checks both
    // `instruments` and `synthetics`, so a synthetic-only entry must be purgeable.
    let mut cache = Cache::default();
    let synth = SyntheticInstrument::default();
    let instrument_id = synth.id;
    cache.add_synthetic(synth).unwrap();

    assert!(cache.synthetic(&instrument_id).is_some());

    cache.purge_instrument(instrument_id);

    assert!(cache.synthetic(&instrument_id).is_none());
    assert!(cache.check_integrity());
}

#[cfg(feature = "defi")]
#[rstest]
fn test_purge_instrument_clears_defi_pools_and_profilers(
    test_pool: Pool,
    test_pool_profiler: PoolProfiler,
) {
    let mut cache = Cache::default();
    let instrument_id = test_pool.instrument_id;
    cache.add_pool(test_pool).unwrap();
    cache.add_pool_profiler(test_pool_profiler).unwrap();

    assert!(cache.pool(&instrument_id).is_some());
    assert!(cache.pool_profiler(&instrument_id).is_some());

    cache.purge_instrument(instrument_id);

    assert!(cache.pool(&instrument_id).is_none());
    assert!(cache.pool_profiler(&instrument_id).is_none());
    assert!(cache.check_integrity());
}

#[rstest]
fn test_purge_closed_positions_does_not_purge_reopened_position() {
    // Create a position that goes FLAT then reopens
    // This test verifies the fix for the race condition where positions that were
    // previously closed but later reopened were incorrectly purged

    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create initial buy order to open position
    let order1 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    // Fill the buy order to open LONG position
    let fill1 = TestOrderEventStubs::filled(
        &order1,
        &audusd_sim,
        Some(TradeId::new("T-1")),            // trade_id
        Some(PositionId::new("P-1")),         // position_id
        Some(Price::from("1.00000")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(1_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    let mut position = Position::new(&audusd_sim, fill1.into());
    let position_id = position.id;

    // Add position to cache
    cache.add_position(&position, OmsType::Netting).unwrap();
    cache.update_position(&position).unwrap();

    // Verify position is LONG
    assert!(position.is_long());
    assert!(!position.is_closed());
    assert!(cache.is_position_open(&position_id));

    // Create sell order to close position (make it FLAT)
    let order2 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .build();

    // Fill the sell order to close position (FLAT)
    let fill2 = TestOrderEventStubs::filled(
        &order2,
        &audusd_sim,
        Some(TradeId::new("T-2")),            // trade_id
        Some(position_id),                    // position_id
        Some(Price::from("1.00010")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(2_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    position.apply(&fill2.into());
    cache.update_position(&position).unwrap();

    // Verify position is now FLAT (closed)
    assert_eq!(position.side, PositionSide::Flat);
    assert!(position.is_closed());
    assert!(position.ts_closed.is_some());
    let ts_closed_original = position.ts_closed.unwrap();
    assert!(cache.is_position_closed(&position_id));

    // Create another buy order to REOPEN the position
    let order3 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(50_000))
        .build();

    // Fill the buy order to reopen position (LONG again)
    let fill3 = TestOrderEventStubs::filled(
        &order3,
        &audusd_sim,
        Some(TradeId::new("T-3")),            // trade_id
        Some(position_id),                    // position_id
        Some(Price::from("1.00020")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(3_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    position.apply(&fill3.into());
    cache.update_position(&position).unwrap();

    // Verify position is LONG again (reopened)
    assert!(position.is_long());
    assert!(!position.is_closed());
    assert_eq!(position.ts_closed, None); // Close timestamp should be reset
    assert!(cache.is_position_open(&position_id));

    // Attempt to purge closed positions
    // This should NOT purge our position even though it was closed before,
    // because it's currently OPEN
    // Use a timestamp far in the future to ensure any old ts_closed would trigger purge
    cache.purge_closed_positions(
        UnixNanos::from(ts_closed_original.as_u64() + 1_000_000_000_000),
        0, // No buffer
    );

    // Position should still exist because it's currently OPEN
    assert!(cache.position_exists(&position_id));
    assert!(cache.position(&position_id).is_some());
    assert!(cache.is_position_open(&position_id));
    assert!(!cache.is_position_closed(&position_id));
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 1);
    assert_eq!(cache.positions_open_count(None, None, None, None, None), 1);
    assert_eq!(
        cache.positions_closed_count(None, None, None, None, None),
        0
    );
}

#[rstest]
fn test_purge_order_cleans_up_strategy_orders_index() {
    // Regression test for strategy_orders index cleanup bug
    // Verifies that after purging an order, it is removed from the strategy's set
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create and add a closed order
    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    let strategy_id = order.strategy_id();
    let client_order_id = order.client_order_id();

    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = OrderEventAny::Accepted(OrderAccepted::default());
    update_order_with_event(&mut cache, &mut order, accepted);

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        None,
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, filled);

    // Verify order is in strategy index
    assert!(cache.index.strategy_orders.contains_key(&strategy_id));
    assert!(
        cache
            .index
            .strategy_orders
            .get(&strategy_id)
            .unwrap()
            .contains(&client_order_id)
    );

    // Purge the order
    cache.purge_order(client_order_id);

    // Verify order is removed from strategy index
    if let Some(strategy_orders) = cache.index.strategy_orders.get(&strategy_id) {
        assert!(!strategy_orders.contains(&client_order_id));
        // If this was the only order, the strategy key should be removed
        assert!(
            !strategy_orders.is_empty(),
            "Empty strategy_orders set should have been removed"
        );
    }

    // Query orders for strategy should not crash and should not include purged order
    let orders_for_strategy = cache.orders(None, None, Some(&strategy_id), None, None);
    assert!(!orders_contains(&orders_for_strategy, &order));
}

#[rstest]
fn test_purge_order_cleans_up_exec_spawn_orders_index() {
    // Regression test for exec_spawn_orders index cleanup bug
    // Verifies that after purging a spawned child order, it is removed from the parent's set
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create parent order
    let parent_id = ClientOrderId::new("PARENT-001");

    // Create and add a child order with exec_spawn_id
    let mut child_order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .exec_spawn_id(parent_id)
        .build();

    let child_id = child_order.client_order_id();

    cache
        .add_order(child_order.clone(), None, None, false)
        .unwrap();

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut child_order, submitted);

    let accepted = OrderEventAny::Accepted(OrderAccepted::default());
    update_order_with_event(&mut cache, &mut child_order, accepted);

    let filled = TestOrderEventStubs::filled(
        &child_order,
        &audusd_sim,
        None,
        None,
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut child_order, filled);

    // Verify child is in parent's spawn set
    assert!(cache.index.exec_spawn_orders.contains_key(&parent_id));
    assert!(
        cache
            .index
            .exec_spawn_orders
            .get(&parent_id)
            .unwrap()
            .contains(&child_id)
    );

    // Purge the child order
    cache.purge_order(child_id);

    // Verify child is removed from parent's spawn set
    if let Some(spawn_orders) = cache.index.exec_spawn_orders.get(&parent_id) {
        assert!(!spawn_orders.contains(&child_id));
    }

    // Query orders for exec spawn should not crash and should not include purged order
    let orders_for_spawn = cache.orders_for_exec_spawn(&parent_id);
    assert!(!orders_contains(&orders_for_spawn, &child_order));
}

#[rstest]
fn test_purge_order_when_order_not_in_cache_still_cleans_up_indices() {
    // Test that even when order is not in cache, indices are cleaned up using forward mapping
    let mut cache = Cache::default();

    let client_order_id = ClientOrderId::new("O-NOT-IN-CACHE");
    let strategy_id = StrategyId::test_default();

    // Manually add to indices (simulating a corrupted state)
    cache
        .index
        .order_strategy
        .insert(client_order_id, strategy_id);
    cache
        .index
        .strategy_orders
        .entry(strategy_id)
        .or_default()
        .insert(client_order_id);

    // Verify indices are set up
    assert!(cache.index.order_strategy.contains_key(&client_order_id));
    assert!(
        cache
            .index
            .strategy_orders
            .get(&strategy_id)
            .unwrap()
            .contains(&client_order_id)
    );

    // Purge order that doesn't exist
    cache.purge_order(client_order_id);

    // Verify indices are cleaned up even though order wasn't in cache
    assert!(!cache.index.order_strategy.contains_key(&client_order_id));
    if let Some(strategy_orders) = cache.index.strategy_orders.get(&strategy_id) {
        assert!(!strategy_orders.contains(&client_order_id));
    }
}

#[rstest]
fn test_purge_order_cleans_up_account_orders_index() {
    // Regression test: purging an order must remove it from account_orders index
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    let client_order_id = order.client_order_id();
    let account_id = AccountId::new("SIM-001");

    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = TestOrderEventStubs::submitted(&order, account_id);
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = TestOrderEventStubs::accepted(&order, account_id, VenueOrderId::new("V-001"));
    update_order_with_event(&mut cache, &mut order, accepted);

    let filled = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        None,
        None,
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, filled);

    // Verify order is in account index (populated by update_order)
    assert!(cache.index.account_orders.contains_key(&account_id));
    assert!(
        cache
            .index
            .account_orders
            .get(&account_id)
            .unwrap()
            .contains(&client_order_id)
    );

    cache.purge_order(client_order_id);

    // Since this was the only order, the account key should be removed entirely
    assert!(!cache.index.account_orders.contains_key(&account_id));

    let orders_for_account = cache.orders(None, None, None, Some(&account_id), None);
    assert!(!orders_contains(&orders_for_account, &order));
}

#[rstest]
fn test_purge_position_cleans_up_account_positions_index() {
    // Regression test: purging a position must remove it from account_positions index
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    let mut order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    let account_id = AccountId::new("SIM-001");
    let trade_id = TradeId::new("T-001");

    cache.add_order(order.clone(), None, None, false).unwrap();

    let submitted = TestOrderEventStubs::submitted(&order, account_id);
    update_order_with_event(&mut cache, &mut order, submitted);

    let accepted = TestOrderEventStubs::accepted(&order, account_id, VenueOrderId::new("V-001"));
    update_order_with_event(&mut cache, &mut order, accepted);

    let filled = TestOrderEventStubs::filled(
        &order,
        &instrument,
        Some(trade_id),
        None,
        Some(Price::from("1.00001")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, filled.clone());

    let position = Position::new(&instrument, filled.into());
    let position_id = position.id;
    cache.add_position(&position, OmsType::Hedging).unwrap();

    // Verify position is in account index (populated by add_position)
    assert!(cache.index.account_positions.contains_key(&account_id));
    assert!(
        cache
            .index
            .account_positions
            .get(&account_id)
            .unwrap()
            .contains(&position_id)
    );

    // Close position so it can be purged (open positions are protected)
    let mut position = cache.position(&position_id).unwrap().clone();
    let close_order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instrument.id())
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-19700101-000000-001-001-2"))
        .build();

    let close_filled = TestOrderEventStubs::filled(
        &close_order,
        &instrument,
        Some(TradeId::new("T-002")),
        Some(position_id),
        Some(Price::from("1.00002")),
        None,
        None,
        None,
        None,
        None,
    );
    let close_filled: OrderFilled = close_filled.into();
    position.apply(&close_filled);
    cache.update_position(&position).unwrap();

    assert!(position.is_closed());

    cache.purge_position(position_id);

    // Since this was the only position, the account key should be removed entirely
    assert!(!cache.index.account_positions.contains_key(&account_id));

    let positions_for_account = cache.positions(None, None, None, Some(&account_id), None);
    assert!(positions_for_account.is_empty());
}

#[rstest]
fn test_update_own_order_book_with_market_order_does_not_panic(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    // Create a LIMIT order to establish an own book for the instrument
    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);
    assert!(cache.own_order_book(&audusd_sim.id()).is_some());

    // Create a MARKET order (no price) and transition it to FILLED
    let market_order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(50_000))
        .client_order_id(ClientOrderId::new("O-19700101-000000-001-001-2"))
        .build();

    cache
        .add_order(market_order.clone(), None, None, false)
        .unwrap();

    let submitted = TestOrderEventStubs::submitted(&market_order, AccountId::new("SIM-001"));
    let mut market_order_mut = market_order;
    market_order_mut.apply(submitted).unwrap();

    let accepted = TestOrderEventStubs::accepted(
        &market_order_mut,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-001"),
    );
    market_order_mut.apply(accepted).unwrap();

    let filled = TestOrderEventStubs::filled(
        &market_order_mut,
        &InstrumentAny::CurrencyPair(audusd_sim.clone()),
        Some(TradeId::new("T-001")),
        None,
        Some(Price::from("1.00010")),
        None,
        None,
        None,
        None,
        None,
    );
    market_order_mut.apply(filled).unwrap();

    // Should not panic - previously would panic at `.expect("OwnBookOrder must have a price")`
    cache.update_own_order_book(&market_order_mut);

    assert!(cache.own_order_book(&audusd_sim.id()).is_some());
}

#[rstest]
fn test_purge_closed_orders_also_purges_order_lists() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    let order_list_id = OrderListId::new("OL-001");

    let mut order1 = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-001"))
        .order_list_id(order_list_id)
        .build();

    let mut order2 = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Sell)
        .price(Price::from("1.00100"))
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-002"))
        .order_list_id(order_list_id)
        .build();
    let order_list = OrderList::new(
        order_list_id,
        instrument.id(),
        order1.strategy_id(),
        vec![order1.client_order_id(), order2.client_order_id()],
        UnixNanos::default(),
    );

    let account_id = AccountId::new("SIM-001");

    cache.add_order(order1.clone(), None, None, false).unwrap();
    cache.add_order(order2.clone(), None, None, false).unwrap();
    cache.add_order_list(order_list).unwrap();

    assert!(cache.order_list_exists(&order_list_id));

    // Transition order1: Initialized -> Submitted -> Accepted -> Filled
    let submitted1 = TestOrderEventStubs::submitted(&order1, account_id);
    update_order_with_event(&mut cache, &mut order1, submitted1);

    let accepted1 = TestOrderEventStubs::accepted(&order1, account_id, VenueOrderId::new("V-001"));
    update_order_with_event(&mut cache, &mut order1, accepted1);

    let filled1 = TestOrderEventStubs::filled(
        &order1,
        &instrument,
        Some(TradeId::new("T-1")),
        None,
        Some(Price::from("1.00000")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order1, filled1);

    // Transition order2: Initialized -> Submitted -> Accepted -> Canceled
    let submitted2 = TestOrderEventStubs::submitted(&order2, account_id);
    update_order_with_event(&mut cache, &mut order2, submitted2);

    let accepted2 = TestOrderEventStubs::accepted(&order2, account_id, VenueOrderId::new("V-002"));
    update_order_with_event(&mut cache, &mut order2, accepted2);

    let canceled2 =
        TestOrderEventStubs::canceled(&order2, account_id, Some(VenueOrderId::new("V-002")));
    update_order_with_event(&mut cache, &mut order2, canceled2);

    assert!(order1.is_closed());
    assert!(order2.is_closed());

    let ts_now = UnixNanos::from(1_000_000_000_000);
    cache.purge_closed_orders(ts_now, 0);

    assert!(!cache.order_exists(&order1.client_order_id()));
    assert!(!cache.order_exists(&order2.client_order_id()));
    assert!(!cache.order_list_exists(&order_list_id));
}

#[rstest]
fn test_purge_closed_orders_does_not_purge_order_list_with_open_orders() {
    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let instrument = InstrumentAny::CurrencyPair(audusd_sim);

    let order_list_id = OrderListId::new("OL-001");

    let mut order1 = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-001"))
        .order_list_id(order_list_id)
        .build();

    let mut order2 = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Sell)
        .price(Price::from("1.00100"))
        .quantity(Quantity::from(100_000))
        .client_order_id(ClientOrderId::new("O-002"))
        .order_list_id(order_list_id)
        .build();
    let order_list = OrderList::new(
        order_list_id,
        instrument.id(),
        order1.strategy_id(),
        vec![order1.client_order_id(), order2.client_order_id()],
        UnixNanos::default(),
    );

    let account_id = AccountId::new("SIM-001");

    cache.add_order(order1.clone(), None, None, false).unwrap();
    cache.add_order(order2.clone(), None, None, false).unwrap();
    cache.add_order_list(order_list).unwrap();

    // Close order1, leave order2 open
    let submitted1 = TestOrderEventStubs::submitted(&order1, account_id);
    update_order_with_event(&mut cache, &mut order1, submitted1);

    let accepted1 = TestOrderEventStubs::accepted(&order1, account_id, VenueOrderId::new("V-001"));
    update_order_with_event(&mut cache, &mut order1, accepted1);

    let filled1 = TestOrderEventStubs::filled(
        &order1,
        &instrument,
        Some(TradeId::new("T-1")),
        None,
        Some(Price::from("1.00000")),
        None,
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order1, filled1);

    let submitted2 = TestOrderEventStubs::submitted(&order2, account_id);
    update_order_with_event(&mut cache, &mut order2, submitted2);

    let accepted2 = TestOrderEventStubs::accepted(&order2, account_id, VenueOrderId::new("V-002"));
    update_order_with_event(&mut cache, &mut order2, accepted2);

    assert!(order1.is_closed());
    assert!(order2.is_open());

    let ts_now = UnixNanos::from(1_000_000_000_000);
    cache.purge_closed_orders(ts_now, 0);

    // Order1 purged, order2 and list remain (order2 still in cache)
    assert!(!cache.order_exists(&order1.client_order_id()));
    assert!(cache.order_exists(&order2.client_order_id()));
    assert!(cache.order_list_exists(&order_list_id));
}

#[rstest]
fn test_force_remove_from_own_order_book(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let submitted = TestOrderEventStubs::submitted(&limit_order, AccountId::new("SIM-001"));
    let mut limit_order_mut = limit_order;
    update_order_with_event(&mut cache, &mut limit_order_mut, submitted);

    assert!(cache.order_exists(&limit_order_mut.client_order_id()));
    assert!(
        cache
            .index
            .orders_inflight
            .contains(&limit_order_mut.client_order_id())
    );
    assert!(cache.own_order_book(&audusd_sim.id()).is_some());

    cache.force_remove_from_own_order_book(&limit_order_mut.client_order_id());

    assert!(
        !cache
            .index
            .orders_open
            .contains(&limit_order_mut.client_order_id())
    );
    assert!(
        !cache
            .index
            .orders_inflight
            .contains(&limit_order_mut.client_order_id())
    );
    assert!(
        !cache
            .index
            .orders_emulated
            .contains(&limit_order_mut.client_order_id())
    );
    assert!(
        !cache
            .index
            .orders_pending_cancel
            .contains(&limit_order_mut.client_order_id())
    );
    assert!(
        cache
            .index
            .orders_closed
            .contains(&limit_order_mut.client_order_id())
    );
}

#[rstest]
fn test_audit_own_order_books_with_inflight_orders(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let submitted = TestOrderEventStubs::submitted(&limit_order, AccountId::new("SIM-001"));
    let mut limit_order_mut = limit_order;
    update_order_with_event(&mut cache, &mut limit_order_mut, submitted);

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    cache.audit_own_order_books();

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(own_book.bids().count() > 0);
}

#[rstest]
fn test_audit_own_order_books_removes_closed(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let submitted = TestOrderEventStubs::submitted(&limit_order, AccountId::new("SIM-001"));
    let mut limit_order_mut = limit_order;
    update_order_with_event(&mut cache, &mut limit_order_mut, submitted);

    let accepted = TestOrderEventStubs::accepted(
        &limit_order_mut,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-001"),
    );
    update_order_with_event(&mut cache, &mut limit_order_mut, accepted);

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    let canceled = TestOrderEventStubs::canceled(
        &limit_order_mut,
        AccountId::new("SIM-001"),
        Some(VenueOrderId::new("V-001")),
    );
    update_order_with_event(&mut cache, &mut limit_order_mut, canceled);

    cache.update_own_order_book(&limit_order_mut);

    cache.audit_own_order_books();

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert_eq!(own_book.bids().count(), 0);
}

#[rstest]
fn test_update_order_removes_closed_ioc_from_existing_own_book(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .time_in_force(TimeInForce::Ioc)
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let mut live_order = limit_order;
    let submitted = TestOrderEventStubs::submitted(&live_order, AccountId::new("SIM-001"));
    update_order_with_event(&mut cache, &mut live_order, submitted);

    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-IOC"),
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    let canceled = TestOrderEventStubs::canceled(
        &live_order,
        AccountId::new("SIM-001"),
        Some(VenueOrderId::new("V-IOC")),
    );
    update_order_with_event(&mut cache, &mut live_order, canceled);

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert_eq!(own_book.bids().count(), 0);
}

#[rstest]
fn test_update_order_venue_id_conflict_still_removes_closed_from_own_book(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let mut live_order = limit_order;
    let submitted = TestOrderEventStubs::submitted(&live_order, AccountId::new("SIM-001"));
    update_order_with_event(&mut cache, &mut live_order, submitted);

    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-ORIGINAL"),
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(own_book.bids().count() > 0);
    assert!(
        cache
            .index
            .orders_open
            .contains(&live_order.client_order_id())
    );

    let filled = OrderFilled::new(
        live_order.trader_id(),
        live_order.strategy_id(),
        audusd_sim.id(),
        live_order.client_order_id(),
        VenueOrderId::new("V-CONFLICT"),
        AccountId::new("SIM-001"),
        TradeId::new("T-CONFLICT"),
        live_order.order_side(),
        live_order.order_type(),
        live_order.quantity(),
        Price::from("1.00000"),
        audusd_sim.quote_currency(),
        LiquiditySide::Maker,
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        false,
        Some(PositionId::new("P-CONFLICT")),
        Some(Money::from("0 USD")),
    );
    update_order_with_event(&mut cache, &mut live_order, OrderEventAny::Filled(filled));

    let own_book = cache.own_order_book(&audusd_sim.id()).unwrap();
    assert!(live_order.is_closed());
    assert_eq!(own_book.bids().count(), 0);
    assert!(
        !cache
            .index
            .orders_open
            .contains(&live_order.client_order_id())
    );
    assert!(
        cache
            .index
            .orders_closed
            .contains(&live_order.client_order_id())
    );
}

#[rstest]
fn test_update_order_allows_venue_id_change_for_order_updated(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();

    let mut live_order = limit_order;
    let submitted = TestOrderEventStubs::submitted(&live_order, AccountId::new("SIM-001"));
    update_order_with_event(&mut cache, &mut live_order, submitted);

    let original_venue_order_id = VenueOrderId::new("V-ORIGINAL");
    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        original_venue_order_id,
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    let new_venue_order_id = VenueOrderId::new("V-UPDATED");
    let updated = OrderEventAny::Updated(OrderUpdated::new(
        live_order.trader_id(),
        live_order.strategy_id(),
        audusd_sim.id(),
        live_order.client_order_id(),
        live_order.quantity(),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        false,
        Some(new_venue_order_id),
        Some(AccountId::new("SIM-001")),
        live_order.price(),
        None,
        None,
        false,
    ));
    update_order_with_event(&mut cache, &mut live_order, updated);

    assert_eq!(live_order.venue_order_id(), Some(new_venue_order_id));
    assert_eq!(
        cache.venue_order_id(&live_order.client_order_id()),
        Some(&new_venue_order_id)
    );
    assert_eq!(
        cache.client_order_id(&new_venue_order_id),
        Some(&live_order.client_order_id())
    );
}

#[rstest]
fn test_update_own_order_book_does_not_create_book_for_closed_order(mut cache: Cache) {
    let audusd_sim = audusd_sim();
    cache
        .add_instrument(InstrumentAny::CurrencyPair(audusd_sim.clone()))
        .unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();

    let mut live_order = limit_order;
    let submitted = TestOrderEventStubs::submitted(&live_order, AccountId::new("SIM-001"));
    update_order_with_event(&mut cache, &mut live_order, submitted);

    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-CLOSED"),
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    let canceled = TestOrderEventStubs::canceled(
        &live_order,
        AccountId::new("SIM-001"),
        Some(VenueOrderId::new("V-CLOSED")),
    );
    update_order_with_event(&mut cache, &mut live_order, canceled);

    assert!(cache.own_order_book(&audusd_sim.id()).is_none());

    cache.update_own_order_book(&live_order);

    assert!(cache.own_order_book(&audusd_sim.id()).is_none());
}

#[rstest]
fn test_own_order_book_lifecycle_sequence(mut cache: Cache) {
    let instrument = InstrumentAny::CurrencyPair(audusd_sim());
    cache.add_instrument(instrument.clone()).unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let mut live_order = limit_order;

    let submitted = TestOrderEventStubs::submitted(&live_order, AccountId::new("SIM-001"));
    update_order_with_event(&mut cache, &mut live_order, submitted);

    let venue_order_id = VenueOrderId::new("V-LCYCLE");
    let accepted =
        TestOrderEventStubs::accepted(&live_order, AccountId::new("SIM-001"), venue_order_id);
    update_order_with_event(&mut cache, &mut live_order, accepted);

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    let partial_fill = TestOrderEventStubs::filled(
        &live_order,
        &instrument,
        None,
        None,
        None,
        Some(Quantity::from(50_000)),
        None,
        None,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut live_order, partial_fill);

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    let canceled = TestOrderEventStubs::canceled(
        &live_order,
        AccountId::new("SIM-001"),
        Some(VenueOrderId::new("V-LCYCLE")),
    );
    update_order_with_event(&mut cache, &mut live_order, canceled);
    cache.update_own_order_book(&live_order);

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert_eq!(own_book.bids().count(), 0);
}

#[rstest]
fn test_own_order_book_pending_cancel_persists_until_final(mut cache: Cache) {
    let instrument = InstrumentAny::CurrencyPair(audusd_sim());
    cache.add_instrument(instrument.clone()).unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let mut live_order = limit_order;
    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-PENDING"),
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    cache.update_order_pending_cancel_local(&live_order);
    cache.audit_own_order_books();

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert!(own_book.bids().count() > 0);

    let canceled = TestOrderEventStubs::canceled(
        &live_order,
        AccountId::new("SIM-001"),
        Some(VenueOrderId::new("V-PENDING")),
    );
    update_order_with_event(&mut cache, &mut live_order, canceled);
    cache.update_own_order_book(&live_order);

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert_eq!(own_book.bids().count(), 0);
}

#[rstest]
fn test_update_own_order_book_reinserts_missing_levels(mut cache: Cache) {
    let instrument = InstrumentAny::CurrencyPair(audusd_sim());
    cache.add_instrument(instrument.clone()).unwrap();

    let limit_order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(instrument.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache
        .add_order(limit_order.clone(), None, None, false)
        .unwrap();
    cache.update_own_order_book(&limit_order);

    let mut live_order = limit_order;
    let accepted = TestOrderEventStubs::accepted(
        &live_order,
        AccountId::new("SIM-001"),
        VenueOrderId::new("V-REINSERT"),
    );
    update_order_with_event(&mut cache, &mut live_order, accepted);

    {
        let own_book = cache
            .own_books
            .get_mut(&instrument.id())
            .expect("own book missing");
        own_book.clear();
    }

    cache.update_own_order_book(&live_order);

    let own_book = cache.own_order_book(&instrument.id()).unwrap();
    assert!(own_book.bids().count() > 0);
}

#[rstest]
fn test_position_flip_netting_mode_cleans_up_closed_index() {
    // Regression test for NETTING position flip index corruption (issue #3081)
    // Verifies that when a position ID is reused in NETTING mode,
    // add_position removes the position from the closed index

    let mut cache = Cache::default();
    let audusd_sim = audusd_sim();
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim);

    // Create initial buy order to open LONG position
    let order1 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();

    // Fill the buy order to open LONG position
    let fill1 = TestOrderEventStubs::filled(
        &order1,
        &audusd_sim,
        Some(TradeId::new("T-1")),            // trade_id
        Some(PositionId::new("P-1")),         // position_id
        Some(Price::from("1.00000")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(1_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    let mut position = Position::new(&audusd_sim, fill1.into());
    let position_id = position.id;

    // Add position to cache
    cache.add_position(&position, OmsType::Netting).unwrap();

    // Verify position is LONG and in open index
    assert!(position.is_long());
    assert!(!position.is_closed());
    assert!(cache.is_position_open(&position_id));
    assert!(!cache.is_position_closed(&position_id));

    // Create a SELL order that closes the position (makes it FLAT)
    let order2 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Sell)
        .quantity(Quantity::from(100_000))
        .build();

    // Fill the sell order to close position to FLAT
    let fill2 = TestOrderEventStubs::filled(
        &order2,
        &audusd_sim,
        Some(TradeId::new("T-2")),            // trade_id
        Some(position_id),                    // position_id (same ID in NETTING)
        Some(Price::from("1.00010")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(2_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    position.apply(&fill2.into());
    cache.update_position(&position).unwrap();

    // Verify position is now FLAT (closed)
    assert_eq!(position.side, PositionSide::Flat);
    assert!(position.is_closed());
    assert!(cache.is_position_closed(&position_id));
    assert!(!cache.is_position_open(&position_id));

    // Snapshot the closed position before reusing the ID (as execution engine does)
    cache.snapshot_position(&position).unwrap();

    // Create a new BUY order to reopen the position (NETTING mode reuses the ID)
    let order3 = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(50_000))
        .build();

    // Fill to create a new LONG position with the same position ID
    let fill3 = TestOrderEventStubs::filled(
        &order3,
        &audusd_sim,
        Some(TradeId::new("T-3")),            // trade_id
        Some(position_id),                    // position_id (reused in NETTING)
        Some(Price::from("1.00020")),         // last_px
        None,                                 // last_qty
        None,                                 // liquidity_side
        None,                                 // commission
        Some(UnixNanos::from(3_000_000_000)), // ts_filled_ns
        None,                                 // account_id
    );

    // Create new position object with the same ID (as execution engine does)
    let position_reopened = Position::new(&audusd_sim, fill3.into());
    assert_eq!(position_reopened.id, position_id); // Same ID reused

    // Add the reopened position to cache
    // THIS IS THE KEY TEST: add_position should remove from closed index
    cache
        .add_position(&position_reopened, OmsType::Netting)
        .unwrap();

    // The reopened position should be in open index, NOT closed index
    assert!(position_reopened.is_long());
    assert!(!position_reopened.is_closed());
    assert!(
        cache.is_position_open(&position_id),
        "Position should be in open index"
    );
    assert!(
        !cache.is_position_closed(&position_id),
        "Position should NOT be in closed index (bug fixed)"
    );

    // Verify position counts
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 1);
    assert_eq!(cache.positions_open_count(None, None, None, None, None), 1);
    assert_eq!(
        cache.positions_closed_count(None, None, None, None, None),
        0
    );

    // Verify the snapshot exists
    assert!(cache.position_snapshots.contains_key(&position_id));

    // Verify the active position is LONG with correct quantity
    let cached_pos = cache.position(&position_id).unwrap();
    assert_eq!(cached_pos.side, PositionSide::Long);
    assert_eq!(cached_pos.quantity, Quantity::from(50_000));
    assert_eq!(cached_pos.event_count(), 1); // Only the reopen fill event
}

#[rstest]
fn test_position_snapshots_round_trip(mut cache: Cache) {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim());

    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let fill = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        Some(TradeId::new("T-1")),
        Some(PositionId::new("P-1")),
        Some(Price::from("1.00000")),
        None,
        None,
        None,
        Some(UnixNanos::from(1_000_000_000)),
        None,
    );
    let position = Position::new(&audusd_sim, fill.into());
    let position_id = position.id;
    let account_id = position.account_id;

    let first_ref = cache.snapshot_position(&position).unwrap();
    cache.snapshot_position(&position).unwrap();
    cache.snapshot_position(&position).unwrap();

    // Frames are stored as one entry per call, not concatenated
    let frames = cache.position_snapshot_bytes(&position_id).unwrap();
    assert_eq!(frames.len(), 3);
    assert_eq!(
        first_ref.blob_ref,
        format!("cache://position-snapshots/{}/0", position_id.as_str()),
    );
    assert_eq!(first_ref.blob.as_ref(), frames[0].as_slice());
    assert_eq!(
        cache
            .load_snapshot_blob(&first_ref.blob_ref)
            .unwrap()
            .unwrap(),
        first_ref.blob,
    );

    // All snapshots round-trip via position_snapshots()
    let snapshots = cache.position_snapshots(Some(&position_id), None);
    assert_eq!(snapshots.len(), 3);

    // Each snapshot has a unique ID derived from the original (UUID suffix)
    let prefix = format!("{}-", position_id.as_str());
    for snapshot in &snapshots {
        assert!(snapshot.id.as_str().starts_with(&prefix));
        assert_ne!(snapshot.id, position_id);
    }
    let unique_ids: AHashSet<_> = snapshots.iter().map(|p| p.id).collect();
    assert_eq!(unique_ids.len(), 3);

    // Account filter keeps matching snapshots
    assert_eq!(cache.position_snapshots(None, Some(&account_id)).len(), 3,);
    // Account filter drops non-matching snapshots
    assert!(
        cache
            .position_snapshots(None, Some(&AccountId::new("OTHER-000")))
            .is_empty(),
    );
}

fn snapshot_test_position() -> Position {
    let audusd_sim = InstrumentAny::CurrencyPair(audusd_sim());
    let order = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(audusd_sim.id())
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .build();
    let fill = TestOrderEventStubs::filled(
        &order,
        &audusd_sim,
        Some(TradeId::new("T-1")),
        Some(PositionId::new("P-1")),
        Some(Price::from("1.00000")),
        None,
        None,
        None,
        Some(UnixNanos::from(1_000_000_000)),
        None,
    );
    Position::new(&audusd_sim, fill.into())
}

#[derive(Default)]
struct SnapshotBlobTestDatabase {
    general: AHashMap<String, Bytes>,
    fail_add: bool,
}

impl SnapshotBlobTestDatabase {
    fn with_general(key: String, value: Bytes) -> Self {
        let mut general = AHashMap::new();
        general.insert(key, value);
        Self {
            general,
            fail_add: false,
        }
    }

    fn fail_add() -> Self {
        Self {
            general: AHashMap::new(),
            fail_add: true,
        }
    }
}

#[async_trait::async_trait]
impl CacheDatabaseAdapter for SnapshotBlobTestDatabase {
    fn close(&mut self) -> anyhow::Result<()> {
        Ok(())
    }

    fn flush(&mut self) -> anyhow::Result<()> {
        Ok(())
    }

    async fn load_all(&self) -> anyhow::Result<CacheMap> {
        Ok(CacheMap::default())
    }

    fn load(&self) -> anyhow::Result<AHashMap<String, Bytes>> {
        Ok(self.general.clone())
    }

    async fn load_currencies(&self) -> anyhow::Result<AHashMap<Ustr, Currency>> {
        Ok(AHashMap::new())
    }

    async fn load_instruments(&self) -> anyhow::Result<AHashMap<InstrumentId, InstrumentAny>> {
        Ok(AHashMap::new())
    }

    async fn load_synthetics(&self) -> anyhow::Result<AHashMap<InstrumentId, SyntheticInstrument>> {
        Ok(AHashMap::new())
    }

    async fn load_accounts(&self) -> anyhow::Result<AHashMap<AccountId, AccountAny>> {
        Ok(AHashMap::new())
    }

    async fn load_orders(&self) -> anyhow::Result<AHashMap<ClientOrderId, OrderAny>> {
        Ok(AHashMap::new())
    }

    async fn load_positions(&self) -> anyhow::Result<AHashMap<PositionId, Position>> {
        Ok(AHashMap::new())
    }

    fn load_index_order_position(&self) -> anyhow::Result<AHashMap<ClientOrderId, Position>> {
        Ok(AHashMap::new())
    }

    fn load_index_order_client(&self) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
        Ok(AHashMap::new())
    }

    async fn load_currency(&self, _code: &Ustr) -> anyhow::Result<Option<Currency>> {
        Ok(None)
    }

    async fn load_instrument(
        &self,
        _instrument_id: &InstrumentId,
    ) -> anyhow::Result<Option<InstrumentAny>> {
        Ok(None)
    }

    async fn load_synthetic(
        &self,
        _instrument_id: &InstrumentId,
    ) -> anyhow::Result<Option<SyntheticInstrument>> {
        Ok(None)
    }

    async fn load_account(&self, _account_id: &AccountId) -> anyhow::Result<Option<AccountAny>> {
        Ok(None)
    }

    async fn load_order(
        &self,
        _client_order_id: &ClientOrderId,
    ) -> anyhow::Result<Option<OrderAny>> {
        Ok(None)
    }

    async fn load_position(&self, _position_id: &PositionId) -> anyhow::Result<Option<Position>> {
        Ok(None)
    }

    fn load_actor(&self, _component_id: &ComponentId) -> anyhow::Result<AHashMap<String, Bytes>> {
        Ok(AHashMap::new())
    }

    fn load_strategy(&self, _strategy_id: &StrategyId) -> anyhow::Result<AHashMap<String, Bytes>> {
        Ok(AHashMap::new())
    }

    fn load_signals(&self, _name: &str) -> anyhow::Result<Vec<Signal>> {
        Ok(Vec::new())
    }

    fn load_custom_data(&self, _data_type: &DataType) -> anyhow::Result<Vec<CustomData>> {
        Ok(Vec::new())
    }

    fn load_order_snapshot(
        &self,
        _client_order_id: &ClientOrderId,
    ) -> anyhow::Result<Option<OrderSnapshot>> {
        Ok(None)
    }

    fn load_position_snapshot(
        &self,
        _position_id: &PositionId,
    ) -> anyhow::Result<Option<PositionSnapshot>> {
        Ok(None)
    }

    fn load_quotes(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<QuoteTick>> {
        Ok(Vec::new())
    }

    fn load_trades(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<TradeTick>> {
        Ok(Vec::new())
    }

    fn load_funding_rates(
        &self,
        _instrument_id: &InstrumentId,
    ) -> anyhow::Result<Vec<FundingRateUpdate>> {
        Ok(Vec::new())
    }

    fn load_bars(&self, _instrument_id: &InstrumentId) -> anyhow::Result<Vec<Bar>> {
        Ok(Vec::new())
    }

    fn add(&self, _key: String, _value: Bytes) -> anyhow::Result<()> {
        if self.fail_add {
            anyhow::bail!("add failed");
        }
        Ok(())
    }

    fn add_currency(&self, _currency: &Currency) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_instrument(&self, _instrument: &InstrumentAny) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_synthetic(&self, _synthetic: &SyntheticInstrument) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_order(&self, _order: &OrderAny, _client_id: Option<ClientId>) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_order_snapshot(&self, _snapshot: &OrderSnapshot) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_position(&self, _position: &Position) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_position_snapshot(&self, _snapshot: &PositionSnapshot) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_order_book(&self, _order_book: &OrderBook) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_signal(&self, _signal: &Signal) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_custom_data(&self, _data: &CustomData) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_quote(&self, _quote: &QuoteTick) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_trade(&self, _trade: &TradeTick) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_funding_rate(&self, _funding_rate: &FundingRateUpdate) -> anyhow::Result<()> {
        Ok(())
    }

    fn add_bar(&self, _bar: &Bar) -> anyhow::Result<()> {
        Ok(())
    }

    fn delete_actor(&self, _component_id: &ComponentId) -> anyhow::Result<()> {
        Ok(())
    }

    fn delete_strategy(&self, _component_id: &StrategyId) -> anyhow::Result<()> {
        Ok(())
    }

    fn delete_order(&self, _client_order_id: &ClientOrderId) -> anyhow::Result<()> {
        Ok(())
    }

    fn delete_position(&self, _position_id: &PositionId) -> anyhow::Result<()> {
        Ok(())
    }

    fn delete_account_event(&self, _account_id: &AccountId, _event_id: &str) -> anyhow::Result<()> {
        Ok(())
    }

    fn index_venue_order_id(
        &self,
        _client_order_id: ClientOrderId,
        _venue_order_id: VenueOrderId,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn index_order_position(
        &self,
        _client_order_id: ClientOrderId,
        _position_id: PositionId,
    ) -> anyhow::Result<()> {
        Ok(())
    }

    fn update_actor(&self) -> anyhow::Result<()> {
        Ok(())
    }

    fn update_strategy(&self) -> anyhow::Result<()> {
        Ok(())
    }

    fn update_account(&self, _account: &AccountAny) -> anyhow::Result<()> {
        Ok(())
    }

    fn update_order(&self, _order_event: &OrderEventAny) -> anyhow::Result<()> {
        Ok(())
    }

    fn update_position(&self, _position: &Position) -> anyhow::Result<()> {
        Ok(())
    }

    fn snapshot_order_state(&self, _order: &OrderAny) -> anyhow::Result<()> {
        Ok(())
    }

    fn snapshot_position_state(&self, _position: &Position) -> anyhow::Result<()> {
        Ok(())
    }

    fn heartbeat(&self, _timestamp: UnixNanos) -> anyhow::Result<()> {
        Ok(())
    }
}

#[rstest]
fn test_restore_position_snapshot_blob(mut cache: Cache) {
    let position = snapshot_test_position();
    let snapshot_ref = cache.snapshot_position(&position).unwrap();
    let mut restored = Cache::default();

    restored
        .restore_snapshot_blob(&snapshot_ref.blob_ref, snapshot_ref.blob.clone())
        .unwrap();
    restored
        .restore_snapshot_blob(&snapshot_ref.blob_ref, snapshot_ref.blob.clone())
        .unwrap();

    let frames = restored.position_snapshot_bytes(&position.id).unwrap();

    assert_eq!(frames.len(), 1);
    assert_eq!(frames[0].as_slice(), snapshot_ref.blob.as_ref());
    assert_eq!(
        restored
            .load_snapshot_blob(&snapshot_ref.blob_ref)
            .unwrap()
            .unwrap(),
        snapshot_ref.blob,
    );
}

#[rstest]
fn test_restore_position_snapshot_blob_rejects_wrong_position(mut cache: Cache) {
    let mut position = snapshot_test_position();
    position.id = PositionId::new("OTHER-POSITION-1");
    let blob = Bytes::from(serde_json::to_vec(&position).unwrap());

    let err = cache
        .restore_snapshot_blob("cache://position-snapshots/P-1/0", blob)
        .unwrap_err();

    assert!(
        err.to_string().contains("does not match blob_ref position"),
        "err was: {err}",
    );
}

#[rstest]
fn test_restore_position_snapshot_blob_rejects_position_id_prefix_collision(mut cache: Cache) {
    let mut source_cache = Cache::default();
    let mut position = snapshot_test_position();
    position.id = PositionId::new("P-1-EXTRA");
    let snapshot_ref = source_cache.snapshot_position(&position).unwrap();

    let err = cache
        .restore_snapshot_blob("cache://position-snapshots/P-1/0", snapshot_ref.blob)
        .unwrap_err();

    assert!(
        err.to_string().contains("does not match blob_ref position"),
        "err was: {err}",
    );
}

#[rstest]
fn test_snapshot_position_failed_persist_does_not_advance_frame_count() {
    let mut cache = Cache::new(None, Some(Box::new(SnapshotBlobTestDatabase::fail_add())));
    let position = snapshot_test_position();

    let err = cache
        .snapshot_position(&position)
        .expect_err("database add failure");

    assert!(err.to_string().contains("add failed"), "err was: {err}");
    assert_eq!(cache.position_snapshot_count(&position.id), 0);
    assert!(cache.position_snapshot_bytes(&position.id).is_none());
}

#[rstest]
fn test_load_snapshot_blob_loads_from_database_when_not_in_memory() {
    let mut source_cache = Cache::default();
    let position = snapshot_test_position();
    let snapshot_ref = source_cache.snapshot_position(&position).unwrap();
    let mut cache = Cache::new(
        None,
        Some(Box::new(SnapshotBlobTestDatabase::with_general(
            snapshot_ref.blob_ref.clone(),
            snapshot_ref.blob.clone(),
        ))),
    );

    let loaded = cache
        .load_snapshot_blob(&snapshot_ref.blob_ref)
        .expect("load snapshot blob");

    assert_eq!(loaded, Some(snapshot_ref.blob));
}

#[rstest]
#[case::unsupported_scheme(
    "file://position-snapshots/P-1/0",
    "unsupported cache snapshot blob_ref"
)]
#[case::missing_frame_separator(
    "cache://position-snapshots/P-1",
    "malformed position snapshot blob_ref"
)]
#[case::empty_position_id("cache://position-snapshots//0", "has empty position id")]
#[case::non_numeric_index(
    "cache://position-snapshots/P-1/not-a-number",
    "has invalid frame index"
)]
fn test_restore_position_snapshot_blob_rejects_malformed_refs(
    #[case] blob_ref: &str,
    #[case] expected: &str,
) {
    let mut source_cache = Cache::default();
    let position = snapshot_test_position();
    let snapshot_ref = source_cache.snapshot_position(&position).unwrap();
    let mut cache = Cache::default();

    let err = cache
        .restore_snapshot_blob(blob_ref, snapshot_ref.blob)
        .expect_err("invalid blob_ref");

    assert!(err.to_string().contains(expected), "err was: {err}");
}

#[rstest]
fn test_restore_position_snapshot_blob_rejects_skipped_frame() {
    let mut source_cache = Cache::default();
    let position = snapshot_test_position();
    let snapshot_ref = source_cache.snapshot_position(&position).unwrap();
    let mut cache = Cache::default();

    let err = cache
        .restore_snapshot_blob("cache://position-snapshots/P-1/1", snapshot_ref.blob)
        .expect_err("skipped frame");

    assert!(
        err.to_string().contains("skips missing frame 0"),
        "err was: {err}",
    );
}

#[rstest]
fn test_restore_position_snapshot_blob_rejects_conflicting_frame_bytes() {
    let mut source_cache = Cache::default();
    let position = snapshot_test_position();
    let first_ref = source_cache.snapshot_position(&position).unwrap();
    let second_ref = source_cache.snapshot_position(&position).unwrap();
    let mut cache = Cache::default();

    cache
        .restore_snapshot_blob(&first_ref.blob_ref, first_ref.blob)
        .expect("restore first frame");
    let err = cache
        .restore_snapshot_blob(&first_ref.blob_ref, second_ref.blob)
        .expect_err("conflicting frame");

    assert!(
        err.to_string()
            .contains("already exists with different bytes"),
        "err was: {err}",
    );
}

#[rstest]
fn test_restore_position_snapshot_blob_rejects_invalid_json() {
    let mut cache = Cache::default();

    let err = cache
        .restore_snapshot_blob(
            "cache://position-snapshots/P-1/0",
            Bytes::from_static(b"not-json"),
        )
        .expect_err("invalid json");

    assert!(err.to_string().contains("expected"), "err was: {err}");
}

#[rstest]
#[case(0)]
#[case(1)]
#[case(3)]
fn test_position_snapshot_count(mut cache: Cache, #[case] n: usize) {
    let position = snapshot_test_position();
    let position_id = position.id;

    for _ in 0..n {
        cache.snapshot_position(&position).unwrap();
    }

    assert_eq!(cache.position_snapshot_count(&position_id), n);
}

#[rstest]
fn test_position_snapshot_count_unknown_position(cache: Cache) {
    assert_eq!(
        cache.position_snapshot_count(&PositionId::new("NOT-PRESENT")),
        0,
    );
}

#[rstest]
fn test_position_snapshots_from_preserves_order_and_skip(mut cache: Cache) {
    let position = snapshot_test_position();
    let position_id = position.id;

    for _ in 0..3 {
        cache.snapshot_position(&position).unwrap();
    }

    // skip=0 returns all three, in insertion order
    let all_from_zero = cache.position_snapshots_from(&position_id, 0);
    assert_eq!(all_from_zero.len(), 3);
    let all_ids: Vec<_> = all_from_zero.iter().map(|p| p.id).collect();

    // skip=1 returns the last two, matching the tail of the full list
    let from_one = cache.position_snapshots_from(&position_id, 1);
    let from_one_ids: Vec<_> = from_one.iter().map(|p| p.id).collect();
    assert_eq!(from_one_ids, all_ids[1..]);

    // skip at or past len returns empty
    assert!(cache.position_snapshots_from(&position_id, 3).is_empty());
    assert!(cache.position_snapshots_from(&position_id, 10).is_empty());

    // Unknown position returns empty regardless of skip
    assert!(
        cache
            .position_snapshots_from(&PositionId::new("NOT-PRESENT"), 0)
            .is_empty(),
    );
}

#[rstest]
fn test_position_snapshots_skip_malformed_frames(mut cache: Cache) {
    let position = snapshot_test_position();
    let position_id = position.id;

    cache.snapshot_position(&position).unwrap();
    // Inject a corrupt frame between two valid ones
    cache
        .position_snapshots
        .get_mut(&position_id)
        .unwrap()
        .push(Bytes::from_static(b"not json"));
    cache.snapshot_position(&position).unwrap();

    // Raw frame count stays authoritative; decoded view drops the bad frame
    assert_eq!(cache.position_snapshot_count(&position_id), 3);
    assert_eq!(
        cache.position_snapshot_bytes(&position_id).unwrap().len(),
        3
    );
    assert_eq!(cache.position_snapshots(Some(&position_id), None).len(), 2);
    assert_eq!(cache.position_snapshots_from(&position_id, 0).len(), 2);
}

#[rstest]
fn test_add_trades_same_timestamp_adds_all(mut cache: Cache) {
    // multiple trades at same timestamp (e.g., large order sweeping levels)
    let ts = UnixNanos::from(1000);
    let instrument_id = InstrumentId::from("AUDUSD.SIM");

    let trade1 = TradeTick::new(
        instrument_id,
        Price::from("1.00000"),
        Quantity::from(100_000),
        AggressorSide::Buyer,
        TradeId::new("1"),
        ts,
        ts,
    );

    let trade2 = TradeTick::new(
        instrument_id,
        Price::from("1.00001"),
        Quantity::from(100_000),
        AggressorSide::Buyer,
        TradeId::new("2"),
        ts,
        ts,
    );

    let trade3 = TradeTick::new(
        instrument_id,
        Price::from("1.00002"),
        Quantity::from(100_000),
        AggressorSide::Buyer,
        TradeId::new("3"),
        ts,
        ts,
    );

    cache.add_trade(trade1).unwrap();
    cache.add_trades(&[trade2, trade3]).unwrap();

    // all three trades should be in cache
    let result = cache.trades(&instrument_id).unwrap();
    assert_eq!(
        result.len(),
        3,
        "All trades with same timestamp should be added"
    );
}

#[rstest]
fn test_add_quotes_same_timestamp_adds_all(mut cache: Cache) {
    // multiple quotes at same timestamp
    let ts = UnixNanos::from(1000);
    let instrument_id = InstrumentId::from("AUDUSD.SIM");

    let quote1 = QuoteTick::new(
        instrument_id,
        Price::from("1.00000"),
        Price::from("1.00001"),
        Quantity::from(100_000),
        Quantity::from(100_000),
        ts,
        ts,
    );

    let quote2 = QuoteTick::new(
        instrument_id,
        Price::from("1.00002"),
        Price::from("1.00003"),
        Quantity::from(100_000),
        Quantity::from(100_000),
        ts,
        ts,
    );

    let quote3 = QuoteTick::new(
        instrument_id,
        Price::from("1.00004"),
        Price::from("1.00005"),
        Quantity::from(100_000),
        Quantity::from(100_000),
        ts,
        ts,
    );

    cache.add_quote(quote1).unwrap();
    cache.add_quotes(&[quote2, quote3]).unwrap();

    // all three quotes should be in cache
    let result = cache.quotes(&instrument_id).unwrap();
    assert_eq!(
        result.len(),
        3,
        "All quotes with same timestamp should be added"
    );
}

#[rstest]
fn test_add_bars_same_timestamp_adds_all(mut cache: Cache) {
    // multiple bars at same timestamp
    let ts = UnixNanos::from(1000);
    let bar_type = BarType::from("AUDUSD.SIM-1-MINUTE-BID-EXTERNAL");

    let bar1 = Bar::new(
        bar_type,
        Price::from("1.00000"),
        Price::from("1.00001"),
        Price::from("0.99999"),
        Price::from("1.00000"),
        Quantity::from(100_000),
        ts,
        ts,
    );

    let bar2 = Bar::new(
        bar_type,
        Price::from("1.00001"),
        Price::from("1.00002"),
        Price::from("1.00000"),
        Price::from("1.00001"),
        Quantity::from(100_000),
        ts,
        ts,
    );

    let bar3 = Bar::new(
        bar_type,
        Price::from("1.00002"),
        Price::from("1.00003"),
        Price::from("1.00001"),
        Price::from("1.00002"),
        Quantity::from(100_000),
        ts,
        ts,
    );

    cache.add_bar(bar1).unwrap();
    cache.add_bars(&[bar2, bar3]).unwrap();

    // all three bars should be in cache
    let result = cache.bars(&bar_type).unwrap();
    assert_eq!(
        result.len(),
        3,
        "All bars with same timestamp should be added"
    );
}

// -- orders_emulated index tests ------------------------------------------------------------------

#[rstest]
fn test_add_emulated_order_indexes_in_orders_emulated(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::StopMarket)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .trigger_price(Price::from("1.00010"))
        .emulation_trigger(TriggerType::LastPrice)
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(
        cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Emulated order should be in orders_emulated index after add"
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 1);
}

#[rstest]
fn test_add_non_emulated_order_not_in_orders_emulated(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .price(Price::from("1.00000"))
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(
        !cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Non-emulated order should not be in orders_emulated index"
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
}

#[rstest]
fn test_initialized_order_indexes_in_orders_active_local(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let mut order = OrderTestBuilder::new(OrderType::Limit)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from(100_000))
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(
        cache
            .index
            .orders_active_local
            .contains(&order.client_order_id()),
        "Initialized order should be in orders_active_local index after add"
    );
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        1
    );

    let submitted = OrderEventAny::Submitted(OrderSubmitted::default());
    update_order_with_event(&mut cache, &mut order, submitted);

    assert!(
        !cache
            .index
            .orders_active_local
            .contains(&order.client_order_id()),
        "Submitted order should be removed from orders_active_local index"
    );
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        0
    );
}

#[rstest]
fn test_released_order_indexes_in_orders_active_local(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::StopMarket)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .trigger_price(Price::from("1.00010"))
        .emulation_trigger(TriggerType::LastPrice)
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    let released = OrderReleased::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        order.client_order_id(),
        Price::from("1.00010"),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
    );
    let mut order = order;
    update_order_with_event(&mut cache, &mut order, OrderEventAny::Released(released));

    assert!(
        cache
            .index
            .orders_active_local
            .contains(&order.client_order_id()),
        "Released order should remain in orders_active_local index"
    );
    assert!(cache.is_order_active_local(&order.client_order_id()));
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        1
    );
}

#[rstest]
fn test_emulated_order_indexes_in_orders_active_local(mut cache: Cache, audusd_sim: CurrencyPair) {
    let order = OrderTestBuilder::new(OrderType::StopMarket)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .trigger_price(Price::from("1.00010"))
        .emulation_trigger(TriggerType::LastPrice)
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    let emulated = OrderEmulated::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        order.client_order_id(),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
    );
    let mut order = order;
    update_order_with_event(&mut cache, &mut order, OrderEventAny::Emulated(emulated));

    assert!(
        cache
            .index
            .orders_active_local
            .contains(&order.client_order_id()),
        "Emulated order should remain in orders_active_local index"
    );
    assert!(cache.is_order_active_local(&order.client_order_id()));
    assert_eq!(
        cache.orders_active_local_count(None, None, None, None, None),
        1
    );
}

#[rstest]
fn test_update_released_order_removes_from_orders_emulated(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let order = OrderTestBuilder::new(OrderType::StopMarket)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .trigger_price(Price::from("1.00010"))
        .emulation_trigger(TriggerType::LastPrice)
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(
        cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Emulated order should be in orders_emulated index after add"
    );

    // Apply released event (order sent to venue, no longer emulated)
    let released = OrderReleased::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        order.client_order_id(),
        Price::from("1.00010"),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
    );
    let mut order = order;
    update_order_with_event(&mut cache, &mut order, OrderEventAny::Released(released));

    assert!(
        !cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Released order should be removed from orders_emulated index"
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
}

#[rstest]
fn test_update_closed_emulated_order_removes_from_orders_emulated(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let order = OrderTestBuilder::new(OrderType::StopMarket)
        .instrument_id(audusd_sim.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from(100_000))
        .trigger_price(Price::from("1.00010"))
        .emulation_trigger(TriggerType::LastPrice)
        .build();

    cache.add_order(order.clone(), None, None, false).unwrap();

    assert!(
        cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Emulated order should be in orders_emulated index after add"
    );

    // Apply emulated event first
    let emulated = OrderEmulated::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        order.client_order_id(),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
    );
    let mut order = order;
    update_order_with_event(&mut cache, &mut order, OrderEventAny::Emulated(emulated));

    // Order should still be emulated
    assert!(
        cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Order should still be in orders_emulated after emulated event"
    );

    // Apply canceled event (order is now closed)
    let canceled = OrderCanceled::new(
        order.trader_id(),
        order.strategy_id(),
        order.instrument_id(),
        order.client_order_id(),
        UUID4::new(),
        UnixNanos::default(),
        UnixNanos::default(),
        false,
        None,
        None,
    );
    update_order_with_event(&mut cache, &mut order, OrderEventAny::Canceled(canceled));

    assert!(
        !cache
            .index
            .orders_emulated
            .contains(&order.client_order_id()),
        "Closed emulated order should be removed from orders_emulated index"
    );
    assert_eq!(cache.orders_emulated_count(None, None, None, None, None), 0);
}

// Covers the size-ordered intersection paths in `query_orders_in_bucket` /
// `count_orders_in_bucket` and the filter-discard regression in `orders_for_exec_algorithm`.

fn build_filter_order(
    instrument_id: InstrumentId,
    side: OrderSide,
    client_order_id: ClientOrderId,
    strategy_id: Option<StrategyId>,
    exec_algorithm_id: Option<ExecAlgorithmId>,
) -> OrderAny {
    let mut builder = OrderTestBuilder::new(OrderType::Limit);
    builder
        .instrument_id(instrument_id)
        .side(side)
        .price(Price::from("1.00000"))
        .quantity(Quantity::from("1"))
        .client_order_id(client_order_id);

    if let Some(strategy_id) = strategy_id {
        builder.strategy_id(strategy_id);
    }

    if let Some(exec_algorithm_id) = exec_algorithm_id {
        builder.exec_algorithm_id(exec_algorithm_id);
    }

    builder.build()
}

fn promote_to_open(
    cache: &mut Cache,
    order: &mut OrderAny,
    account_id: AccountId,
    venue_order_id: VenueOrderId,
) {
    let submitted = TestOrderEventStubs::submitted(order, account_id);
    update_order_with_event(cache, order, submitted);
    let accepted = TestOrderEventStubs::accepted(order, account_id, venue_order_id);
    update_order_with_event(cache, order, accepted);
}

fn order_id_set(orders: &[OrderRef<'_>]) -> AHashSet<ClientOrderId> {
    orders.iter().map(|o| o.client_order_id()).collect()
}

#[rstest]
fn test_orders_for_exec_algorithm_applies_filters(mut cache: Cache) {
    // Mixed universe: TWAP and VWAP across two venues / two instruments / two sides.
    // The filter discard regression in the original Rust port returned every TWAP order
    // regardless of venue/instrument/side; this test pins the corrected behavior.
    let venue_a = Venue::from("VENUE-A");
    let inst_a1 = InstrumentId::from("SYMBOL-1.VENUE-A");
    let inst_a2 = InstrumentId::from("SYMBOL-2.VENUE-A");
    let inst_b1 = InstrumentId::from("SYMBOL-1.VENUE-B");
    let twap = ExecAlgorithmId::from("TWAP");
    let vwap = ExecAlgorithmId::from("VWAP");

    let twap_a1_buy = build_filter_order(
        inst_a1,
        OrderSide::Buy,
        ClientOrderId::from("O-T-A1-BUY"),
        None,
        Some(twap),
    );
    let twap_a2_sell = build_filter_order(
        inst_a2,
        OrderSide::Sell,
        ClientOrderId::from("O-T-A2-SELL"),
        None,
        Some(twap),
    );
    let twap_b1_buy = build_filter_order(
        inst_b1,
        OrderSide::Buy,
        ClientOrderId::from("O-T-B1-BUY"),
        None,
        Some(twap),
    );
    let vwap_a1_buy = build_filter_order(
        inst_a1,
        OrderSide::Buy,
        ClientOrderId::from("O-V-A1-BUY"),
        None,
        Some(vwap),
    );
    let untagged_a1 = build_filter_order(
        inst_a1,
        OrderSide::Buy,
        ClientOrderId::from("O-UN-A1"),
        None,
        None,
    );

    for order in [
        &twap_a1_buy,
        &twap_a2_sell,
        &twap_b1_buy,
        &vwap_a1_buy,
        &untagged_a1,
    ] {
        cache.add_order(order.clone(), None, None, false).unwrap();
    }

    // No filter: all TWAP orders, regardless of venue or instrument
    assert_eq!(
        order_id_set(&cache.orders_for_exec_algorithm(&twap, None, None, None, None, None)),
        [
            twap_a1_buy.client_order_id(),
            twap_a2_sell.client_order_id(),
            twap_b1_buy.client_order_id(),
        ]
        .into_iter()
        .collect::<AHashSet<_>>()
    );

    // Venue filter: only VENUE-A TWAP orders
    assert_eq!(
        order_id_set(&cache.orders_for_exec_algorithm(
            &twap,
            Some(&venue_a),
            None,
            None,
            None,
            None,
        )),
        [
            twap_a1_buy.client_order_id(),
            twap_a2_sell.client_order_id()
        ]
        .into_iter()
        .collect::<AHashSet<_>>()
    );

    // Venue + instrument filter: pinpoints a single TWAP order
    assert_eq!(
        order_id_set(&cache.orders_for_exec_algorithm(
            &twap,
            Some(&venue_a),
            Some(&inst_a1),
            None,
            None,
            None,
        )),
        [twap_a1_buy.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );

    // Side filter (Buy) excludes the Sell-side TWAP order
    assert_eq!(
        order_id_set(&cache.orders_for_exec_algorithm(
            &twap,
            None,
            None,
            None,
            None,
            Some(OrderSide::Buy),
        )),
        [twap_a1_buy.client_order_id(), twap_b1_buy.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );

    // VWAP id sees only the VWAP-tagged order
    assert_eq!(
        order_id_set(&cache.orders_for_exec_algorithm(&vwap, None, None, None, None, None)),
        [vwap_a1_buy.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );
}

#[rstest]
fn test_orders_for_exec_algorithm_unknown_id_returns_empty(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    let twap = ExecAlgorithmId::from("TWAP");
    let unknown = ExecAlgorithmId::from("UNKNOWN");
    let order = build_filter_order(
        audusd_sim.id,
        OrderSide::Buy,
        ClientOrderId::from("O-1"),
        None,
        Some(twap),
    );
    cache.add_order(order, None, None, false).unwrap();

    assert!(
        cache
            .orders_for_exec_algorithm(&unknown, None, None, None, None, None)
            .is_empty()
    );
}

#[rstest]
fn test_orders_for_exec_algorithm_unknown_venue_returns_empty(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    // Drives the FilterSources::Empty path inside query_orders_in_bucket: a filter is
    // provided but its index entry is missing, so the result must be empty regardless
    // of what is in the bucket.
    let twap = ExecAlgorithmId::from("TWAP");
    let other_venue = Venue::from("OTHER-VENUE");
    let order = build_filter_order(
        audusd_sim.id,
        OrderSide::Buy,
        ClientOrderId::from("O-1"),
        None,
        Some(twap),
    );
    cache.add_order(order, None, None, false).unwrap();

    assert!(
        cache
            .orders_for_exec_algorithm(&twap, Some(&other_venue), None, None, None, None)
            .is_empty()
    );
}

#[rstest]
fn test_client_order_ids_state_buckets_with_filters(mut cache: Cache) {
    // Universe: 4 orders across 2 venues × 2 instruments. Two are promoted to OPEN and
    // two stay INITIALIZED (active_local). Filter combinations exercise the size-ordered
    // intersection in `query_orders_in_bucket` against state-specific buckets, not just
    // the universal `index.orders` bucket exercised by `test_client_order_ids_filtering`.
    let venue_a = Venue::from("VENUE-A");
    let venue_b = Venue::from("VENUE-B");
    let inst_a = InstrumentId::from("SYMBOL-1.VENUE-A");
    let inst_b = InstrumentId::from("SYMBOL-1.VENUE-B");
    let account_id = AccountId::from("SIM-001");

    let mut o_a_open = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-A-OPEN"),
        None,
        None,
    );
    let mut o_b_open = build_filter_order(
        inst_b,
        OrderSide::Sell,
        ClientOrderId::from("O-B-OPEN"),
        None,
        None,
    );
    let o_a_init = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-A-INIT"),
        None,
        None,
    );
    let o_b_init = build_filter_order(
        inst_b,
        OrderSide::Sell,
        ClientOrderId::from("O-B-INIT"),
        None,
        None,
    );

    cache
        .add_order(o_a_open.clone(), None, None, false)
        .unwrap();
    cache
        .add_order(o_b_open.clone(), None, None, false)
        .unwrap();
    cache
        .add_order(o_a_init.clone(), None, None, false)
        .unwrap();
    cache
        .add_order(o_b_init.clone(), None, None, false)
        .unwrap();

    promote_to_open(
        &mut cache,
        &mut o_a_open,
        account_id,
        VenueOrderId::from("V-A-1"),
    );
    promote_to_open(
        &mut cache,
        &mut o_b_open,
        account_id,
        VenueOrderId::from("V-B-1"),
    );

    // No filter: open bucket has both promoted orders, active_local has both initialized
    assert_eq!(
        cache.client_order_ids_open(None, None, None, None),
        [o_a_open.client_order_id(), o_b_open.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );
    assert_eq!(
        cache.client_order_ids_active_local(None, None, None, None),
        [o_a_init.client_order_id(), o_b_init.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );

    // Venue filter: bucket fold restricts to the requested venue's slice of each state
    assert_eq!(
        cache.client_order_ids_open(Some(&venue_a), None, None, None),
        [o_a_open.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );
    assert_eq!(
        cache.client_order_ids_active_local(Some(&venue_b), None, None, None),
        [o_b_init.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );

    // Venue + instrument filter: same instrument in different states must not bleed
    // across buckets
    assert_eq!(
        cache.client_order_ids_open(Some(&venue_a), Some(&inst_a), None, None),
        [o_a_open.client_order_id()]
            .into_iter()
            .collect::<AHashSet<_>>()
    );
    assert!(
        cache
            .client_order_ids_open(Some(&venue_a), Some(&inst_b), None, None)
            .is_empty()
    );

    // Closed bucket has nothing yet
    assert!(
        cache
            .client_order_ids_closed(None, None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .client_order_ids_closed(Some(&venue_a), None, None, None)
            .is_empty()
    );
}

#[rstest]
fn test_orders_count_with_filters_and_side(mut cache: Cache) {
    // Drives `count_orders_in_bucket` through the Sets arm with side filtering. Existing
    // count tests in this file exercise only the no-filter no-side path.
    let venue_a = Venue::from("VENUE-A");
    let venue_b = Venue::from("VENUE-B");
    let inst_a = InstrumentId::from("SYMBOL-1.VENUE-A");
    let inst_b = InstrumentId::from("SYMBOL-1.VENUE-B");
    let strategy = StrategyId::from("S-FILTER-001");
    let account_id = AccountId::from("SIM-001");

    let mut buy_a = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-BUY-A"),
        Some(strategy),
        None,
    );
    let mut sell_a = build_filter_order(
        inst_a,
        OrderSide::Sell,
        ClientOrderId::from("O-SELL-A"),
        Some(strategy),
        None,
    );
    let mut buy_b = build_filter_order(
        inst_b,
        OrderSide::Buy,
        ClientOrderId::from("O-BUY-B"),
        Some(strategy),
        None,
    );

    cache.add_order(buy_a.clone(), None, None, false).unwrap();
    cache.add_order(sell_a.clone(), None, None, false).unwrap();
    cache.add_order(buy_b.clone(), None, None, false).unwrap();

    promote_to_open(
        &mut cache,
        &mut buy_a,
        account_id,
        VenueOrderId::from("V-1"),
    );
    promote_to_open(
        &mut cache,
        &mut sell_a,
        account_id,
        VenueOrderId::from("V-2"),
    );
    promote_to_open(
        &mut cache,
        &mut buy_b,
        account_id,
        VenueOrderId::from("V-3"),
    );

    // No filter, no side
    assert_eq!(cache.orders_open_count(None, None, None, None, None), 3);

    // Venue filter
    assert_eq!(
        cache.orders_open_count(Some(&venue_a), None, None, None, None),
        2
    );
    assert_eq!(
        cache.orders_open_count(Some(&venue_b), None, None, None, None),
        1
    );

    // Venue + instrument filter
    assert_eq!(
        cache.orders_open_count(Some(&venue_a), Some(&inst_a), None, None, None),
        2
    );

    // Strategy filter (all orders share the strategy)
    assert_eq!(
        cache.orders_open_count(None, None, Some(&strategy), None, None),
        3
    );

    // Side filter alone (Unfiltered + side branch)
    assert_eq!(
        cache.orders_open_count(None, None, None, None, Some(OrderSide::Buy)),
        2
    );
    assert_eq!(
        cache.orders_open_count(None, None, None, None, Some(OrderSide::Sell)),
        1
    );

    // Venue + side (Sets + side branch)
    assert_eq!(
        cache.orders_open_count(Some(&venue_a), None, None, None, Some(OrderSide::Buy)),
        1
    );
    assert_eq!(
        cache.orders_open_count(Some(&venue_a), None, None, None, Some(OrderSide::Sell)),
        1
    );
    assert_eq!(
        cache.orders_open_count(Some(&venue_b), None, None, None, Some(OrderSide::Sell)),
        0
    );

    // Total count with side filter applies across all buckets the same way
    assert_eq!(
        cache.orders_total_count(None, None, None, None, Some(OrderSide::Buy)),
        2
    );

    // Unknown venue: FilterSources::Empty path returns 0
    let unknown = Venue::from("OTHER");
    assert_eq!(
        cache.orders_open_count(Some(&unknown), None, None, None, None),
        0
    );
    assert_eq!(
        cache.orders_open_count(Some(&unknown), Some(&inst_a), None, None, None),
        0
    );
}

#[rstest]
fn test_unknown_filter_returns_empty_across_query_methods(
    mut cache: Cache,
    audusd_sim: CurrencyPair,
) {
    // Pins the FilterSources::Empty path: when a filter argument names a key that does
    // not appear in the corresponding index, the result is unconditionally empty.
    let order = build_filter_order(
        audusd_sim.id,
        OrderSide::Buy,
        ClientOrderId::from("O-1"),
        None,
        None,
    );
    cache.add_order(order, None, None, false).unwrap();

    let unknown_venue = Venue::from("OTHER-VENUE");
    let unknown_instrument = InstrumentId::from("SYMBOL-NONE.NOWHERE");
    let unknown_strategy = StrategyId::from("S-UNKNOWN");

    assert!(
        cache
            .client_order_ids(Some(&unknown_venue), None, None, None)
            .is_empty()
    );
    assert!(
        cache
            .client_order_ids(None, Some(&unknown_instrument), None, None)
            .is_empty()
    );
    assert!(
        cache
            .client_order_ids(None, None, Some(&unknown_strategy), None)
            .is_empty()
    );
    assert!(
        cache
            .client_order_ids_active_local(Some(&unknown_venue), None, None, None)
            .is_empty()
    );
    assert_eq!(
        cache.orders_total_count(Some(&unknown_venue), None, None, None, None),
        0
    );
    assert_eq!(
        cache.orders_active_local_count(Some(&unknown_venue), None, None, None, None),
        0
    );
}

#[rstest]
fn test_position_filters_with_state_and_side(mut cache: Cache) {
    // Mirrors the order coverage for positions. Builds two open positions on different
    // venues and a closed position on venue A; asserts filter and side branches against
    // `position_*_ids` and `positions_*_count`.
    fn make_pair(id_str: &str) -> CurrencyPair {
        CurrencyPair::new(
            InstrumentId::from(id_str),
            Symbol::from(id_str),
            Currency::USD(),
            Currency::EUR(),
            2,
            4,
            Price::from("0.01"),
            Quantity::from("0.0001"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    let venue_a = Venue::from("VENUE-A");
    let venue_b = Venue::from("VENUE-B");
    let instr_a = make_pair("PAIR-1.VENUE-A");
    let instr_b = make_pair("PAIR-1.VENUE-B");

    let order_a = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_a.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from("1"))
        .build();
    let fill_a = match TestOrderEventStubs::filled(
        &order_a,
        &InstrumentAny::CurrencyPair(instr_a.clone()),
        None,
        Some(PositionId::new("POS-A")),
        None,
        None,
        None,
        None,
        None,
        None,
    ) {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let pos_a_long = Position::new(&InstrumentAny::CurrencyPair(instr_a), fill_a);

    let order_b = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_b.id)
        .side(OrderSide::Sell)
        .quantity(Quantity::from("1"))
        .build();
    let fill_b = match TestOrderEventStubs::filled(
        &order_b,
        &InstrumentAny::CurrencyPair(instr_b.clone()),
        None,
        Some(PositionId::new("POS-B")),
        None,
        None,
        None,
        None,
        None,
        None,
    ) {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let mut pos_b_short = Position::new(&InstrumentAny::CurrencyPair(instr_b), fill_b);
    pos_b_short.side = PositionSide::Short;

    let mut pos_a_closed = pos_a_long.clone();
    pos_a_closed.id = PositionId::new("POS-A-CLOSED");
    pos_a_closed.side = PositionSide::Flat;
    pos_a_closed.ts_closed = Some(UnixNanos::from(1));

    cache.add_position(&pos_a_long, OmsType::Netting).unwrap();
    cache.add_position(&pos_b_short, OmsType::Netting).unwrap();
    cache.add_position(&pos_a_closed, OmsType::Netting).unwrap();
    // `add_position` always inserts into `positions_open`; the closed flag is honored
    // only when state is reconciled via `update_position`.
    cache.update_position(&pos_a_closed).unwrap();

    // Open ids: filter by venue
    assert_eq!(
        cache.position_open_ids(Some(&venue_a), None, None, None),
        [pos_a_long.id].into_iter().collect::<AHashSet<_>>()
    );
    assert_eq!(
        cache.position_open_ids(Some(&venue_b), None, None, None),
        [pos_b_short.id].into_iter().collect::<AHashSet<_>>()
    );

    // Closed ids: only the flagged position on venue A
    assert_eq!(
        cache.position_closed_ids(None, None, None, None),
        [pos_a_closed.id].into_iter().collect::<AHashSet<_>>()
    );
    assert_eq!(
        cache.position_closed_ids(Some(&venue_a), None, None, None),
        [pos_a_closed.id].into_iter().collect::<AHashSet<_>>()
    );
    assert!(
        cache
            .position_closed_ids(Some(&venue_b), None, None, None)
            .is_empty()
    );

    // Counts with filters (side-filter coverage is symmetric with `count_orders_in_bucket`
    // and is exercised by `test_orders_count_with_filters_and_side`).
    assert_eq!(cache.positions_open_count(None, None, None, None, None), 2);
    assert_eq!(
        cache.positions_open_count(Some(&venue_a), None, None, None, None),
        1
    );
    assert_eq!(cache.positions_total_count(None, None, None, None, None), 3);
    assert_eq!(
        cache.positions_closed_count(None, None, None, None, None),
        1
    );

    // Unknown filter -> empty / 0
    let unknown = Venue::from("OTHER");
    assert!(
        cache
            .position_open_ids(Some(&unknown), None, None, None)
            .is_empty()
    );
    assert_eq!(
        cache.positions_open_count(Some(&unknown), None, None, None, None),
        0
    );
}

// Pins the semantic invariants of the new `has_*`, `_view`, and `iter_*` query API
// families against the existing owned/`_count` methods. A wrong-bucket binding or
// inverted intersection in any wrapper would break one of these consistency checks.

fn populate_orders_universe(
    cache: &mut Cache,
) -> (Venue, Venue, InstrumentId, InstrumentId, StrategyId) {
    let venue_a = Venue::from("VENUE-A");
    let venue_b = Venue::from("VENUE-B");
    let inst_a = InstrumentId::from("SYMBOL-1.VENUE-A");
    let inst_b = InstrumentId::from("SYMBOL-1.VENUE-B");
    let strategy = StrategyId::from("S-CONS-001");
    let account_id = AccountId::from("SIM-001");

    let mut o_a_open = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-A-OPEN"),
        Some(strategy),
        None,
    );
    let mut o_b_open = build_filter_order(
        inst_b,
        OrderSide::Sell,
        ClientOrderId::from("O-B-OPEN"),
        Some(strategy),
        None,
    );
    let o_a_init = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-A-INIT"),
        Some(strategy),
        None,
    );
    let o_b_init = build_filter_order(
        inst_b,
        OrderSide::Sell,
        ClientOrderId::from("O-B-INIT"),
        Some(strategy),
        None,
    );

    cache
        .add_order(o_a_open.clone(), None, None, false)
        .unwrap();
    cache
        .add_order(o_b_open.clone(), None, None, false)
        .unwrap();
    cache.add_order(o_a_init, None, None, false).unwrap();
    cache.add_order(o_b_init, None, None, false).unwrap();

    promote_to_open(
        cache,
        &mut o_a_open,
        account_id,
        VenueOrderId::from("V-A-1"),
    );
    promote_to_open(
        cache,
        &mut o_b_open,
        account_id,
        VenueOrderId::from("V-B-1"),
    );

    (venue_a, venue_b, inst_a, inst_b, strategy)
}

fn assert_orders_apis_consistent(
    cache: &Cache,
    venue: Option<&Venue>,
    instrument: Option<&InstrumentId>,
    strategy: Option<&StrategyId>,
) {
    macro_rules! check_bucket {
        ($owned:ident, $view:ident, $iter:ident, $count:ident, $has:ident, $label:literal) => {{
            let owned = cache.$owned(venue, instrument, strategy, None);
            let view = cache.$view(venue, instrument, strategy, None);
            let iter: AHashSet<ClientOrderId> =
                cache.$iter(venue, instrument, strategy, None).collect();
            let count = cache.$count(venue, instrument, strategy, None, None);
            let has = cache.$has(venue, instrument, strategy, None, None);

            assert_eq!(
                view.as_ref(),
                &owned,
                "view != owned for {} / {venue:?} / {instrument:?} / {strategy:?}",
                $label,
            );
            assert_eq!(
                iter, owned,
                "iter.collect != owned for {} / {venue:?} / {instrument:?} / {strategy:?}",
                $label,
            );
            assert_eq!(
                count,
                owned.len(),
                "count != owned.len() for {} / {venue:?} / {instrument:?} / {strategy:?}",
                $label,
            );
            assert_eq!(
                has,
                !owned.is_empty(),
                "has != !owned.is_empty() for {} / {venue:?} / {instrument:?} / {strategy:?}",
                $label,
            );
        }};
    }

    check_bucket!(
        client_order_ids,
        client_order_ids_view,
        iter_client_order_ids,
        orders_total_count,
        has_orders,
        "all"
    );
    check_bucket!(
        client_order_ids_open,
        client_order_ids_open_view,
        iter_client_order_ids_open,
        orders_open_count,
        has_orders_open,
        "open"
    );
    check_bucket!(
        client_order_ids_closed,
        client_order_ids_closed_view,
        iter_client_order_ids_closed,
        orders_closed_count,
        has_orders_closed,
        "closed"
    );
    check_bucket!(
        client_order_ids_active_local,
        client_order_ids_active_local_view,
        iter_client_order_ids_active_local,
        orders_active_local_count,
        has_orders_active_local,
        "active_local"
    );
    check_bucket!(
        client_order_ids_emulated,
        client_order_ids_emulated_view,
        iter_client_order_ids_emulated,
        orders_emulated_count,
        has_orders_emulated,
        "emulated"
    );
    check_bucket!(
        client_order_ids_inflight,
        client_order_ids_inflight_view,
        iter_client_order_ids_inflight,
        orders_inflight_count,
        has_orders_inflight,
        "inflight"
    );
}

#[rstest]
fn test_orders_query_apis_are_consistent(mut cache: Cache) {
    let (venue_a, venue_b, inst_a, inst_b, strategy) = populate_orders_universe(&mut cache);

    let combos: [(Option<&Venue>, Option<&InstrumentId>, Option<&StrategyId>); 7] = [
        (None, None, None),
        (Some(&venue_a), None, None),
        (Some(&venue_b), None, None),
        (None, Some(&inst_a), None),
        (Some(&venue_a), Some(&inst_a), None),
        (Some(&venue_a), Some(&inst_b), None),
        (None, None, Some(&strategy)),
    ];

    for (venue, instrument, strategy_filter) in combos {
        assert_orders_apis_consistent(&cache, venue, instrument, strategy_filter);
    }
}

fn assert_positions_apis_consistent(
    cache: &Cache,
    venue: Option<&Venue>,
    instrument: Option<&InstrumentId>,
    strategy: Option<&StrategyId>,
) {
    macro_rules! check_bucket {
        ($owned:ident, $view:ident, $iter:ident, $count:ident, $has:ident, $label:literal) => {{
            let owned = cache.$owned(venue, instrument, strategy, None);
            let view = cache.$view(venue, instrument, strategy, None);
            let iter: AHashSet<PositionId> =
                cache.$iter(venue, instrument, strategy, None).collect();
            let count = cache.$count(venue, instrument, strategy, None, None);
            let has = cache.$has(venue, instrument, strategy, None, None);

            assert_eq!(view.as_ref(), &owned, "view != owned for {}", $label);
            assert_eq!(iter, owned, "iter.collect != owned for {}", $label);
            assert_eq!(count, owned.len(), "count != owned.len() for {}", $label);
            assert_eq!(
                has,
                !owned.is_empty(),
                "has != !owned.is_empty() for {}",
                $label,
            );
        }};
    }

    check_bucket!(
        position_ids,
        position_ids_view,
        iter_position_ids,
        positions_total_count,
        has_positions,
        "all"
    );
    check_bucket!(
        position_open_ids,
        position_open_ids_view,
        iter_position_open_ids,
        positions_open_count,
        has_positions_open,
        "open"
    );
    check_bucket!(
        position_closed_ids,
        position_closed_ids_view,
        iter_position_closed_ids,
        positions_closed_count,
        has_positions_closed,
        "closed"
    );
}

#[rstest]
fn test_positions_query_apis_are_consistent(mut cache: Cache) {
    fn make_pair(id_str: &str) -> CurrencyPair {
        CurrencyPair::new(
            InstrumentId::from(id_str),
            Symbol::from(id_str),
            Currency::USD(),
            Currency::EUR(),
            2,
            4,
            Price::from("0.01"),
            Quantity::from("0.0001"),
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            None,
            UnixNanos::default(),
            UnixNanos::default(),
        )
    }

    let venue_a = Venue::from("VENUE-A");
    let venue_b = Venue::from("VENUE-B");
    let instr_a = make_pair("PAIR-1.VENUE-A");
    let instr_b = make_pair("PAIR-1.VENUE-B");

    let order_a = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_a.id)
        .side(OrderSide::Buy)
        .quantity(Quantity::from("1"))
        .build();
    let fill_a = match TestOrderEventStubs::filled(
        &order_a,
        &InstrumentAny::CurrencyPair(instr_a.clone()),
        None,
        Some(PositionId::new("POS-A")),
        None,
        None,
        None,
        None,
        None,
        None,
    ) {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let pos_a = Position::new(&InstrumentAny::CurrencyPair(instr_a), fill_a);

    let order_b = OrderTestBuilder::new(OrderType::Market)
        .instrument_id(instr_b.id)
        .side(OrderSide::Sell)
        .quantity(Quantity::from("1"))
        .build();
    let fill_b = match TestOrderEventStubs::filled(
        &order_b,
        &InstrumentAny::CurrencyPair(instr_b.clone()),
        None,
        Some(PositionId::new("POS-B")),
        None,
        None,
        None,
        None,
        None,
        None,
    ) {
        OrderEventAny::Filled(f) => f,
        _ => unreachable!(),
    };
    let pos_b = Position::new(&InstrumentAny::CurrencyPair(instr_b), fill_b);

    let mut pos_a_closed = pos_a.clone();
    pos_a_closed.id = PositionId::new("POS-A-CLOSED");
    pos_a_closed.side = PositionSide::Flat;
    pos_a_closed.ts_closed = Some(UnixNanos::from(1));

    cache.add_position(&pos_a, OmsType::Netting).unwrap();
    cache.add_position(&pos_b, OmsType::Netting).unwrap();
    cache.add_position(&pos_a_closed, OmsType::Netting).unwrap();
    cache.update_position(&pos_a_closed).unwrap();

    let combos: [(Option<&Venue>, Option<&InstrumentId>); 5] = [
        (None, None),
        (Some(&venue_a), None),
        (Some(&venue_b), None),
        (Some(&venue_a), Some(&pos_a.instrument_id)),
        (None, Some(&pos_b.instrument_id)),
    ];

    for (venue, instrument) in combos {
        assert_positions_apis_consistent(&cache, venue, instrument, None);
    }
}

#[rstest]
fn test_has_orders_with_side_filter(mut cache: Cache) {
    // Mixed-side universe so side filtering can flip results
    let venue_a = Venue::from("VENUE-A");
    let inst_a = InstrumentId::from("SYMBOL-1.VENUE-A");
    let strategy = StrategyId::from("S-SIDE-001");
    let account_id = AccountId::from("SIM-001");

    let mut buy = build_filter_order(
        inst_a,
        OrderSide::Buy,
        ClientOrderId::from("O-BUY"),
        Some(strategy),
        None,
    );
    let mut sell = build_filter_order(
        inst_a,
        OrderSide::Sell,
        ClientOrderId::from("O-SELL"),
        Some(strategy),
        None,
    );
    cache.add_order(buy.clone(), None, None, false).unwrap();
    cache.add_order(sell.clone(), None, None, false).unwrap();
    promote_to_open(&mut cache, &mut buy, account_id, VenueOrderId::from("V-1"));
    promote_to_open(&mut cache, &mut sell, account_id, VenueOrderId::from("V-2"));

    let cases: [(Option<&Venue>, Option<OrderSide>, bool); 6] = [
        (None, None, true),
        (None, Some(OrderSide::NoOrderSide), true),
        (None, Some(OrderSide::Buy), true),
        (None, Some(OrderSide::Sell), true),
        (Some(&venue_a), Some(OrderSide::Buy), true),
        (Some(&Venue::from("OTHER")), Some(OrderSide::Buy), false),
    ];

    for (venue, side, expected) in cases {
        assert_eq!(
            cache.has_orders_open(venue, None, None, None, side),
            expected,
            "has_orders_open mismatch for venue={venue:?} side={side:?}",
        );
        assert_eq!(
            cache.has_orders_open(venue, None, None, None, side),
            cache.orders_open_count(venue, None, None, None, side) > 0,
            "has_orders_open != count > 0 for venue={venue:?} side={side:?}",
        );
        assert_eq!(
            cache.has_orders_active_local(venue, None, None, None, side),
            cache.orders_active_local_count(venue, None, None, None, side) > 0,
            "has_orders_active_local != count > 0 for venue={venue:?} side={side:?}",
        );
    }
}

#[rstest]
fn test_unknown_filter_consistent_across_new_apis(mut cache: Cache, audusd_sim: CurrencyPair) {
    // Same fixture shape as the existing unknown-filter test, but pinned against the new
    // `has_*`, `_view`, and `iter_*` API families to lock the FilterSources::Empty path.
    let order = build_filter_order(
        audusd_sim.id,
        OrderSide::Buy,
        ClientOrderId::from("O-1"),
        None,
        None,
    );
    cache.add_order(order, None, None, false).unwrap();

    let unknown_venue = Venue::from("OTHER-VENUE");
    let unknown_instrument = InstrumentId::from("SYMBOL-NONE.NOWHERE");
    let unknown_strategy = StrategyId::from("S-UNKNOWN");

    // has_*: should all be false
    assert!(!cache.has_orders(Some(&unknown_venue), None, None, None, None));
    assert!(!cache.has_orders_active_local(Some(&unknown_venue), None, None, None, None));
    assert!(!cache.has_orders_active_local(None, Some(&unknown_instrument), None, None, None));
    assert!(!cache.has_orders_active_local(None, None, Some(&unknown_strategy), None, None));

    // _view: should be Cow::Owned(empty)
    let view_venue = cache.client_order_ids_view(Some(&unknown_venue), None, None, None);
    assert!(view_venue.is_empty());
    assert!(matches!(view_venue, Cow::Owned(_)));
    let view_active =
        cache.client_order_ids_active_local_view(Some(&unknown_venue), None, None, None);
    assert!(view_active.is_empty());
    assert!(matches!(view_active, Cow::Owned(_)));

    // iter_*: should be empty
    assert_eq!(
        cache
            .iter_client_order_ids(Some(&unknown_venue), None, None, None)
            .count(),
        0,
    );
    assert_eq!(
        cache
            .iter_client_order_ids_active_local(Some(&unknown_venue), None, None, None)
            .count(),
        0,
    );
}

#[rstest]
fn test_view_returns_borrowed_when_unfiltered(mut cache: Cache, audusd_sim: CurrencyPair) {
    // Loaded so the bucket is non-empty; the borrow check is independent of contents
    let order = build_filter_order(
        audusd_sim.id,
        OrderSide::Buy,
        ClientOrderId::from("O-1"),
        None,
        None,
    );
    cache.add_order(order, None, None, false).unwrap();

    // Each unfiltered view must return a borrow that points at the corresponding index entry
    macro_rules! check_borrow {
        ($view:ident, $bucket:ident) => {{
            let view = cache.$view(None, None, None, None);
            match &view {
                Cow::Borrowed(set) => assert!(
                    std::ptr::eq(*set, &cache.index.$bucket),
                    "{} should borrow from index.{}",
                    stringify!($view),
                    stringify!($bucket),
                ),
                Cow::Owned(_) => panic!(
                    "{} should return Cow::Borrowed when unfiltered",
                    stringify!($view),
                ),
            }
        }};
    }

    check_borrow!(client_order_ids_view, orders);
    check_borrow!(client_order_ids_open_view, orders_open);
    check_borrow!(client_order_ids_closed_view, orders_closed);
    check_borrow!(client_order_ids_active_local_view, orders_active_local);
    check_borrow!(client_order_ids_emulated_view, orders_emulated);
    check_borrow!(client_order_ids_inflight_view, orders_inflight);
    check_borrow!(position_ids_view, positions);
    check_borrow!(position_open_ids_view, positions_open);
    check_borrow!(position_closed_ids_view, positions_closed);
}