nautilus-execution 0.64.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.
// -------------------------------------------------------------------------------------------------

//! Base execution client functionality.

use std::sync::atomic::{AtomicBool, Ordering};

use nautilus_common::cache::{Cache, CacheView};
use nautilus_model::{
    enums::{AccountType, OmsType},
    identifiers::{AccountId, ClientId, ClientOrderId, TraderId, Venue},
    orders::{OrderAny, OrderList},
    types::Currency,
};

/// Base implementation for execution clients providing identity and connection state.
///
/// This struct provides the foundation for all execution clients, holding
/// client identity, connection state, and read-only cache access. Execution
/// clients use this as a base and extend it with venue-specific implementations.
///
/// For event generation, use `OrderEventFactory` from `nautilus_common::factories`.
/// For live adapters, use `ExecutionEventEmitter` which combines event generation
/// with async dispatch. For backtest/sandbox, use `OrderEventFactory` directly
/// and dispatch via `msgbus::send_order_event()`.
#[derive(Debug)]
pub struct ExecutionClientCore {
    pub trader_id: TraderId,
    pub client_id: ClientId,
    pub venue: Venue,
    pub oms_type: OmsType,
    pub account_id: AccountId,
    pub account_type: AccountType,
    pub base_currency: Option<Currency>,
    connected: AtomicBool,
    started: AtomicBool,
    instruments_initialized: AtomicBool,
    cache: CacheView,
}

impl Clone for ExecutionClientCore {
    fn clone(&self) -> Self {
        Self {
            trader_id: self.trader_id,
            client_id: self.client_id,
            venue: self.venue,
            oms_type: self.oms_type,
            account_id: self.account_id,
            account_type: self.account_type,
            base_currency: self.base_currency,
            connected: AtomicBool::new(self.connected.load(Ordering::Acquire)),
            started: AtomicBool::new(self.started.load(Ordering::Acquire)),
            instruments_initialized: AtomicBool::new(
                self.instruments_initialized.load(Ordering::Acquire),
            ),
            cache: self.cache.clone(),
        }
    }
}

impl ExecutionClientCore {
    /// Creates a new [`ExecutionClientCore`] instance.
    #[expect(clippy::too_many_arguments)]
    #[must_use]
    pub fn new(
        trader_id: TraderId,
        client_id: ClientId,
        venue: Venue,
        oms_type: OmsType,
        account_id: AccountId,
        account_type: AccountType,
        base_currency: Option<Currency>,
        cache: impl Into<CacheView>,
    ) -> Self {
        Self {
            trader_id,
            client_id,
            venue,
            oms_type,
            account_id,
            account_type,
            base_currency,
            connected: AtomicBool::new(false),
            started: AtomicBool::new(false),
            instruments_initialized: AtomicBool::new(false),
            cache: cache.into(),
        }
    }

    /// Returns a read-only borrow of the cache.
    pub fn cache(&self) -> std::cell::Ref<'_, Cache> {
        self.cache.borrow()
    }

    /// Returns the order for the given `client_order_id` from the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if the order is not found in the cache.
    pub fn get_order(&self, client_order_id: &ClientOrderId) -> anyhow::Result<OrderAny> {
        Ok(self.cache.borrow().try_order_owned(client_order_id)?)
    }

    /// Returns all orders for the given order list from the cache.
    ///
    /// # Errors
    ///
    /// Returns an error if any order is not found in the cache.
    pub fn get_orders_for_list(&self, order_list: &OrderList) -> anyhow::Result<Vec<OrderAny>> {
        order_list
            .client_order_ids
            .iter()
            .map(|id| self.get_order(id))
            .collect()
    }

    /// Returns `true` if the client is connected.
    #[must_use]
    pub fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Acquire)
    }

    /// Returns `true` if the client is disconnected.
    #[must_use]
    pub fn is_disconnected(&self) -> bool {
        !self.is_connected()
    }

    /// Sets the client as connected.
    pub fn set_connected(&self) {
        self.connected.store(true, Ordering::Release);
    }

    /// Sets the client as disconnected.
    pub fn set_disconnected(&self) {
        self.connected.store(false, Ordering::Release);
    }

    /// Returns `true` if the client has been started.
    #[must_use]
    pub fn is_started(&self) -> bool {
        self.started.load(Ordering::Acquire)
    }

    /// Returns `true` if the client has not been started.
    #[must_use]
    pub fn is_stopped(&self) -> bool {
        !self.is_started()
    }

    /// Sets the client as started.
    pub fn set_started(&self) {
        self.started.store(true, Ordering::Release);
    }

    /// Sets the client as stopped.
    pub fn set_stopped(&self) {
        self.started.store(false, Ordering::Release);
    }

    /// Returns `true` if instruments have been initialized.
    #[must_use]
    pub fn instruments_initialized(&self) -> bool {
        self.instruments_initialized.load(Ordering::Acquire)
    }

    /// Sets instruments as initialized.
    pub fn set_instruments_initialized(&self) {
        self.instruments_initialized.store(true, Ordering::Release);
    }

    /// Sets the account identifier for the execution client.
    pub const fn set_account_id(&mut self, account_id: AccountId) {
        self.account_id = account_id;
    }
}

#[cfg(test)]
mod tests {
    use std::{cell::RefCell, rc::Rc};

    use nautilus_common::cache::OrderLookupError;
    use nautilus_core::UnixNanos;
    use nautilus_model::{
        enums::{OrderSide, OrderType},
        identifiers::OrderListId,
        orders::{Order, builder::OrderTestBuilder},
        types::{Price, Quantity},
    };
    use rstest::rstest;

    use super::*;

    #[rstest]
    fn test_get_orders_for_list_preserves_order_and_cached_fields() {
        let first = OrderTestBuilder::new(OrderType::Limit)
            .client_order_id(ClientOrderId::from("O-SECOND"))
            .instrument_id("AUD/USD.SIM".into())
            .side(OrderSide::Buy)
            .price(Price::from("0.65001"))
            .quantity(Quantity::from(17))
            .build();
        let second = OrderTestBuilder::new(OrderType::Limit)
            .client_order_id(ClientOrderId::from("O-FIRST"))
            .instrument_id("EUR/USD.SIM".into())
            .side(OrderSide::Sell)
            .price(Price::from("1.08002"))
            .quantity(Quantity::from(29))
            .build();
        let cache = Rc::new(RefCell::new(Cache::default()));

        for order in [&second, &first] {
            cache
                .borrow_mut()
                .add_order(order.clone(), None, None, false)
                .unwrap();
        }

        let core = core(cache);
        let list = OrderList::new(
            OrderListId::from("OL-001"),
            first.instrument_id(),
            first.strategy_id(),
            vec![first.client_order_id(), second.client_order_id()],
            UnixNanos::new(123),
        );

        let orders = core.get_orders_for_list(&list).unwrap();

        assert_eq!(orders.len(), 2);
        for (actual, expected) in orders.iter().zip([&first, &second]) {
            // OrderAny equality compares only client_order_id
            assert_eq!(actual.init_event(), expected.init_event());
            assert_eq!(actual.status(), expected.status());
            assert_eq!(actual.filled_qty(), expected.filled_qty());
        }
    }

    #[rstest]
    #[case(0)]
    #[case(1)]
    fn test_get_orders_for_list_rejects_missing_member(#[case] missing_index: usize) {
        let order = OrderTestBuilder::new(OrderType::Market)
            .client_order_id(ClientOrderId::from("O-PRESENT"))
            .instrument_id("AUD/USD.SIM".into())
            .side(OrderSide::Buy)
            .quantity(Quantity::from(17))
            .build();
        let cache = Rc::new(RefCell::new(Cache::default()));
        cache
            .borrow_mut()
            .add_order(order.clone(), None, None, false)
            .unwrap();
        let core = core(cache);
        let missing_id = ClientOrderId::from("O-MISSING");
        let mut ids = vec![order.client_order_id()];
        ids.insert(missing_index, missing_id);
        let list = OrderList::new(
            OrderListId::from("OL-001"),
            order.instrument_id(),
            order.strategy_id(),
            ids,
            UnixNanos::new(123),
        );

        let error = core.get_orders_for_list(&list).unwrap_err();

        assert_eq!(
            error.downcast_ref::<OrderLookupError>(),
            Some(&OrderLookupError::NotFound {
                client_order_id: missing_id
            }),
        );
        assert_eq!(
            core.get_order(&order.client_order_id())
                .unwrap()
                .init_event(),
            order.init_event()
        );
    }

    fn core(cache: Rc<RefCell<Cache>>) -> ExecutionClientCore {
        ExecutionClientCore::new(
            TraderId::from("TRADER-007"),
            ClientId::from("CLIENT-003"),
            Venue::from("SIM"),
            OmsType::Hedging,
            AccountId::from("SIM-009"),
            AccountType::Margin,
            Some(Currency::USD()),
            cache,
        )
    }
}