nautilus_bitmex/http/

client.rs

// -------------------------------------------------------------------------------------------------
//  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.
// -------------------------------------------------------------------------------------------------

//! Provides the HTTP client integration for the [BitMEX](https://bitmex.com) REST API.
//!
//! This module defines and implements a [`BitmexHttpClient`] for
//! sending requests to various BitMEX endpoints. It handles request signing
//! (when credentials are provided), constructs valid HTTP requests
//! using the [`HttpClient`], and parses the responses back into structured data or a [`BitmexHttpError`].
//!
//! BitMEX API reference <https://www.bitmex.com/api/explorer/#/default>.

use std::{
    collections::HashMap,
    num::NonZeroU32,
    sync::{
        Arc, LazyLock, RwLock,
        atomic::{AtomicBool, Ordering},
    },
};

use chrono::{DateTime, Timelike, Utc};
use dashmap::DashMap;
use nautilus_common::cache::InstrumentLookupError;
use nautilus_core::{
    AtomicMap, AtomicTime, UUID4, UnixNanos,
    consts::{NAUTILUS_TRADER, NAUTILUS_USER_AGENT},
    env::get_or_env_var_opt,
    time::get_atomic_clock_realtime,
};
use nautilus_model::{
    data::{
        Bar, BarType, BookOrder, FundingRateUpdate, OrderBookDelta, OrderBookDeltas, TradeTick,
    },
    enums::{
        AccountType, AggregationSource, BarAggregation, BookAction, BookType, ContingencyType,
        OrderSide, OrderType, PriceType, RecordFlag, TimeInForce, TrailingOffsetType, TriggerType,
    },
    events::AccountState,
    identifiers::{AccountId, ClientOrderId, InstrumentId, OrderListId, VenueOrderId},
    instruments::{Instrument as InstrumentTrait, InstrumentAny},
    orderbook::OrderBook,
    reports::{FillReport, OrderStatusReport, PositionStatusReport},
    types::{MarginBalance, Money, Price, Quantity},
};
use nautilus_network::{
    http::{HttpClient, Method, StatusCode, USER_AGENT},
    ratelimiter::quota::Quota,
    retry::{RetryConfig, RetryManager},
};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use serde_json::Value;
use tokio_util::sync::CancellationToken;
use ustr::Ustr;

use super::{
    error::{BitmexErrorResponse, BitmexHttpError},
    models::{
        BitmexApiInfo, BitmexExecution, BitmexFunding, BitmexInstrument, BitmexMargin, BitmexOrder,
        BitmexOrderBookL2, BitmexPosition, BitmexTrade, BitmexTradeBin, BitmexWallet,
    },
    query::{
        DeleteAllOrdersParams, DeleteOrderParams, GetExecutionParams, GetExecutionParamsBuilder,
        GetFundingParams, GetFundingParamsBuilder, GetOrderBookL2Params,
        GetOrderBookL2ParamsBuilder, GetOrderParams, GetPositionParams, GetPositionParamsBuilder,
        GetTradeBucketedParams, GetTradeBucketedParamsBuilder, GetTradeParams,
        GetTradeParamsBuilder, PostCancelAllAfterParams, PostOrderParams,
        PostPositionLeverageParams, PutOrderParams,
    },
};
use crate::{
    common::{
        consts::{BITMEX_HTTP_TESTNET_URL, BITMEX_HTTP_URL},
        credential::{Credential, credential_env_vars},
        enums::{
            BitmexContingencyType, BitmexEnvironment, BitmexExecInstruction, BitmexOrderStatus,
            BitmexOrderType, BitmexPegPriceType, BitmexSide, BitmexTimeInForce,
        },
        parse::{
            bitmex_account_id, bitmex_currency_divisor, parse_account_balance,
            parse_contracts_quantity, quantity_to_u32,
        },
    },
    http::{
        parse::{
            InstrumentParseResult, parse_fill_report, parse_instrument_any,
            parse_order_status_report, parse_position_report, parse_trade, parse_trade_bin,
        },
        query::{DeleteAllOrdersParamsBuilder, GetOrderParamsBuilder, PutOrderParamsBuilder},
    },
    websocket::messages::BitmexMarginMsg,
};

/// Default BitMEX REST API rate limits.
///
/// BitMEX implements a dual-layer rate limiting system:
/// - Primary limit: 120 requests per minute for authenticated users (30 for unauthenticated).
/// - Secondary limit: 10 requests per second burst limit for specific endpoints.
const BITMEX_DEFAULT_RATE_LIMIT_PER_SECOND: u32 = 10;
const BITMEX_DEFAULT_RATE_LIMIT_PER_MINUTE_AUTHENTICATED: u32 = 120;
const BITMEX_DEFAULT_RATE_LIMIT_PER_MINUTE_UNAUTHENTICATED: u32 = 30;
const BITMEX_MAX_TABLE_COUNT: u32 = 500;

const BITMEX_GLOBAL_RATE_KEY: &str = "bitmex:global";
const BITMEX_MINUTE_RATE_KEY: &str = "bitmex:minute";

static RATE_LIMIT_KEYS: LazyLock<Vec<Ustr>> = LazyLock::new(|| {
    vec![
        Ustr::from(BITMEX_GLOBAL_RATE_KEY),
        Ustr::from(BITMEX_MINUTE_RATE_KEY),
    ]
});

/// Represents a BitMEX HTTP response.
#[derive(Debug, Serialize, Deserialize)]
pub struct BitmexResponse<T> {
    /// The typed data returned by the BitMEX endpoint.
    pub data: Vec<T>,
}

/// Provides a lower-level HTTP client for connecting to the [BitMEX](https://bitmex.com) REST API.
///
/// This client wraps the underlying [`HttpClient`] to handle functionality
/// specific to BitMEX, such as request signing (for authenticated endpoints),
/// forming request URLs, and deserializing responses into specific data models.
///
/// # Connection Management
///
/// The client uses HTTP keep-alive for connection pooling with a 90-second idle timeout,
/// which matches BitMEX's server-side keep-alive timeout. Connections are automatically
/// reused for subsequent requests to minimize latency.
///
/// # Rate Limiting
///
/// BitMEX enforces the following rate limits:
/// - 120 requests per minute for authenticated users (30 for unauthenticated).
/// - 10 requests per second burst limit for certain endpoints (order management).
///
/// The client automatically respects these limits through the configured quota.
#[derive(Debug, Clone)]
pub struct BitmexRawHttpClient {
    base_url: String,
    client: HttpClient,
    credential: Option<Credential>,
    recv_window_ms: u64,
    retry_manager: RetryManager<BitmexHttpError>,
    cancellation_token: Arc<RwLock<CancellationToken>>,
}

impl Default for BitmexRawHttpClient {
    fn default() -> Self {
        Self::new(
            None,
            60,
            3,
            1000,
            10_000,
            10_000,
            BITMEX_DEFAULT_RATE_LIMIT_PER_SECOND,
            BITMEX_DEFAULT_RATE_LIMIT_PER_MINUTE_UNAUTHENTICATED,
            None,
        )
        .expect("Failed to create default BitmexHttpInnerClient")
    }
}

impl BitmexRawHttpClient {
    /// Creates a new [`BitmexRawHttpClient`] using the default BitMEX HTTP URL,
    /// optionally overridden with a custom base URL.
    ///
    /// This version of the client has **no credentials**, so it can only
    /// call publicly accessible endpoints.
    ///
    /// # Errors
    ///
    /// Returns an error if the retry manager cannot be created.
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        base_url: Option<String>,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        recv_window_ms: u64,
        max_requests_per_second: u32,
        max_requests_per_minute: u32,
        proxy_url: Option<String>,
    ) -> Result<Self, BitmexHttpError> {
        let retry_config = RetryConfig {
            max_retries,
            initial_delay_ms: retry_delay_ms,
            max_delay_ms: retry_delay_max_ms,
            backoff_factor: 2.0,
            jitter_ms: 1000,
            operation_timeout_ms: Some(60_000),
            immediate_first: false,
            max_elapsed_ms: Some(180_000),
        };

        let retry_manager = RetryManager::new(retry_config);

        Ok(Self {
            base_url: base_url.unwrap_or(BITMEX_HTTP_URL.to_string()),
            client: HttpClient::new(
                Self::default_headers(),
                vec![],
                Self::rate_limiter_quotas(max_requests_per_second, max_requests_per_minute)?,
                Some(Self::default_quota(max_requests_per_second)?),
                Some(timeout_secs),
                proxy_url,
            )
            .map_err(|e| {
                BitmexHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
            })?,
            credential: None,
            recv_window_ms,
            retry_manager,
            cancellation_token: Arc::new(RwLock::new(CancellationToken::new())),
        })
    }

    /// Creates a new [`BitmexRawHttpClient`] configured with credentials
    /// for authenticated requests, optionally using a custom base URL.
    ///
    /// # Errors
    ///
    /// Returns an error if the retry manager cannot be created.
    #[expect(clippy::too_many_arguments)]
    pub fn with_credentials(
        api_key: String,
        api_secret: String,
        base_url: String,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        recv_window_ms: u64,
        max_requests_per_second: u32,
        max_requests_per_minute: u32,
        proxy_url: Option<String>,
    ) -> Result<Self, BitmexHttpError> {
        let retry_config = RetryConfig {
            max_retries,
            initial_delay_ms: retry_delay_ms,
            max_delay_ms: retry_delay_max_ms,
            backoff_factor: 2.0,
            jitter_ms: 1000,
            operation_timeout_ms: Some(60_000),
            immediate_first: false,
            max_elapsed_ms: Some(180_000),
        };

        let retry_manager = RetryManager::new(retry_config);

        Ok(Self {
            base_url,
            client: HttpClient::new(
                Self::default_headers(),
                vec![],
                Self::rate_limiter_quotas(max_requests_per_second, max_requests_per_minute)?,
                Some(Self::default_quota(max_requests_per_second)?),
                Some(timeout_secs),
                proxy_url,
            )
            .map_err(|e| {
                BitmexHttpError::NetworkError(format!("Failed to create HTTP client: {e}"))
            })?,
            credential: Some(Credential::new(api_key, api_secret)),
            recv_window_ms,
            retry_manager,
            cancellation_token: Arc::new(RwLock::new(CancellationToken::new())),
        })
    }

    fn default_headers() -> HashMap<String, String> {
        HashMap::from([(USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string())])
    }

    fn default_quota(max_requests_per_second: u32) -> Result<Quota, BitmexHttpError> {
        let burst = NonZeroU32::new(max_requests_per_second)
            .unwrap_or(NonZeroU32::new(BITMEX_DEFAULT_RATE_LIMIT_PER_SECOND).expect("non-zero"));
        Quota::per_second(burst).ok_or_else(|| {
            BitmexHttpError::ValidationError(format!(
                "Invalid max_requests_per_second: {max_requests_per_second} exceeds maximum"
            ))
        })
    }

    fn rate_limiter_quotas(
        max_requests_per_second: u32,
        max_requests_per_minute: u32,
    ) -> Result<Vec<(String, Quota)>, BitmexHttpError> {
        let per_sec_quota = Self::default_quota(max_requests_per_second)?;
        let per_min_quota =
            Quota::per_minute(NonZeroU32::new(max_requests_per_minute).unwrap_or_else(|| {
                NonZeroU32::new(BITMEX_DEFAULT_RATE_LIMIT_PER_MINUTE_AUTHENTICATED)
                    .expect("non-zero")
            }));

        Ok(vec![
            (BITMEX_GLOBAL_RATE_KEY.to_string(), per_sec_quota),
            (BITMEX_MINUTE_RATE_KEY.to_string(), per_min_quota),
        ])
    }

    fn rate_limit_keys() -> Vec<Ustr> {
        RATE_LIMIT_KEYS.clone()
    }

    /// Cancel all pending HTTP requests.
    ///
    /// # Panics
    ///
    /// Panics if the cancellation token lock is poisoned.
    pub fn cancel_all_requests(&self) {
        self.cancellation_token
            .read()
            .expect("cancellation token lock poisoned")
            .cancel();
    }

    /// Replace the cancellation token so new requests can proceed.
    ///
    /// # Panics
    ///
    /// Panics if the cancellation token lock is poisoned.
    pub fn reset_cancellation_token(&self) {
        *self
            .cancellation_token
            .write()
            .expect("cancellation token lock poisoned") = CancellationToken::new();
    }

    /// Get a clone of the cancellation token for this client.
    ///
    /// # Panics
    ///
    /// Panics if the cancellation token lock is poisoned.
    pub fn cancellation_token(&self) -> CancellationToken {
        self.cancellation_token
            .read()
            .expect("cancellation token lock poisoned")
            .clone()
    }

    fn sign_request(
        &self,
        method: &Method,
        endpoint: &str,
        body: Option<&[u8]>,
    ) -> Result<HashMap<String, String>, BitmexHttpError> {
        let credential = self
            .credential
            .as_ref()
            .ok_or(BitmexHttpError::MissingCredentials)?;

        let expires = Utc::now().timestamp() + (self.recv_window_ms / 1000) as i64;
        let body_str = body.and_then(|b| std::str::from_utf8(b).ok()).unwrap_or("");

        let full_path = if endpoint.starts_with("/api/v1") {
            endpoint.to_string()
        } else {
            format!("/api/v1{endpoint}")
        };

        let signature = credential.sign(method.as_str(), &full_path, expires, body_str);

        let mut headers = HashMap::new();
        headers.insert("api-expires".to_string(), expires.to_string());
        headers.insert("api-key".to_string(), credential.api_key().to_string());
        headers.insert("api-signature".to_string(), signature);

        // Add Content-Type header for form-encoded body
        if body.is_some()
            && (*method == Method::POST || *method == Method::PUT || *method == Method::DELETE)
        {
            headers.insert(
                "Content-Type".to_string(),
                "application/x-www-form-urlencoded".to_string(),
            );
        }

        Ok(headers)
    }

    async fn send_request<T: DeserializeOwned, P: Serialize>(
        &self,
        method: Method,
        endpoint: &str,
        params: Option<&P>,
        body: Option<Vec<u8>>,
        authenticate: bool,
    ) -> Result<T, BitmexHttpError> {
        let endpoint = endpoint.to_string();
        let method_clone = method.clone();
        let body_clone = body.clone();

        // Serialize params before closure to avoid reference lifetime issues
        // Query params are used with GET and DELETE methods
        let params_str = if method == Method::GET || method == Method::DELETE {
            params
                .map(serde_urlencoded::to_string)
                .transpose()
                .map_err(|e| {
                    BitmexHttpError::JsonError(format!("Failed to serialize params: {e}"))
                })?
        } else {
            None
        };

        let full_endpoint = match params_str {
            Some(ref query) if !query.is_empty() => format!("{endpoint}?{query}"),
            _ => endpoint.clone(),
        };

        let url = format!("{}{}", self.base_url, full_endpoint);

        let operation = || {
            let url = url.clone();
            let method = method_clone.clone();
            let body = body_clone.clone();
            let full_endpoint = full_endpoint.clone();

            async move {
                let headers = if authenticate {
                    Some(self.sign_request(&method, &full_endpoint, body.as_deref())?)
                } else {
                    None
                };

                let rate_keys = Self::rate_limit_keys();
                let resp = self
                    .client
                    .request_with_ustr_keys(method, url, None, headers, body, None, Some(rate_keys))
                    .await?;

                if resp.status.is_success() {
                    serde_json::from_slice(&resp.body).map_err(Into::into)
                } else if let Ok(error_resp) =
                    serde_json::from_slice::<BitmexErrorResponse>(&resp.body)
                {
                    Err(error_resp.into())
                } else {
                    Err(BitmexHttpError::UnexpectedStatus {
                        status: StatusCode::from_u16(resp.status.as_u16())
                            .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
                        body: String::from_utf8_lossy(&resp.body).to_string(),
                    })
                }
            }
        };

        // Retry strategy based on BitMEX error responses and HTTP status codes:
        //
        // 1. Network errors: always retry (transient connection issues).
        // 2. HTTP 5xx/429: server errors and rate limiting should be retried.
        // 3. BitMEX JSON errors with specific handling:
        //    - "RateLimitError": explicit rate limit error from BitMEX.
        //    - "HTTPError": generic error name used by BitMEX for various issues
        //      Only retry if message contains "rate limit" to avoid retrying
        //      non-transient errors like authentication failures, validation errors,
        //      insufficient balance, etc. which also return as "HTTPError".
        //
        // Note: BitMEX returns many permanent errors as "HTTPError" (e.g., "Invalid orderQty",
        // "Account has insufficient Available Balance", "Invalid API Key") which should NOT
        // be retried. We only retry when the message explicitly mentions rate limiting.
        //
        // See tests in tests/http.rs for retry behavior validation.
        let should_retry = |error: &BitmexHttpError| -> bool {
            match error {
                BitmexHttpError::NetworkError(_) => true,
                BitmexHttpError::UnexpectedStatus { status, .. } => {
                    status.as_u16() >= 500 || status.as_u16() == 429
                }
                BitmexHttpError::BitmexError {
                    error_name,
                    message,
                } => {
                    error_name == "RateLimitError"
                        || (error_name == "HTTPError"
                            && message.to_lowercase().contains("rate limit"))
                }
                _ => false,
            }
        };

        let create_error = |msg: String| -> BitmexHttpError {
            if msg == "canceled" {
                BitmexHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
            } else {
                BitmexHttpError::NetworkError(msg)
            }
        };

        let cancel_token = self.cancellation_token();

        self.retry_manager
            .execute_with_retry_with_cancel(
                endpoint.as_str(),
                operation,
                should_retry,
                create_error,
                &cancel_token,
            )
            .await
    }

    /// Get all instruments.
    ///
    /// Instruments that cannot be deserialized (e.g. unknown fields for new BitMEX
    /// instrument types) are skipped with a warning rather than failing the whole
    /// response.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the response is not a JSON array.
    pub async fn get_instruments(
        &self,
        active_only: bool,
    ) -> Result<Vec<BitmexInstrument>, BitmexHttpError> {
        let path = if active_only {
            "/instrument/active"
        } else {
            "/instrument"
        };
        let raw: Vec<serde_json::Value> = self
            .send_request::<_, ()>(Method::GET, path, None, None, false)
            .await?;

        let raw_len = raw.len();
        let mut instruments = Vec::with_capacity(raw_len);

        for value in raw {
            match serde_json::from_value::<BitmexInstrument>(value) {
                Ok(inst) => instruments.push(inst),
                Err(e) => {
                    log::warn!("Skipping instrument that could not be deserialized: {e}");
                }
            }
        }

        if raw_len > 0 && instruments.is_empty() {
            return Err(BitmexHttpError::JsonError(format!(
                "All {raw_len} instrument(s) failed to deserialize; venue schema may have changed"
            )));
        }

        Ok(instruments)
    }

    /// Requests the current server time from BitMEX.
    ///
    /// Retrieves the BitMEX API info including the system time in Unix timestamp (milliseconds).
    /// This is useful for synchronizing local clocks with the exchange server and logging time drift.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or if the response body
    /// cannot be parsed into [`BitmexApiInfo`].
    pub async fn get_server_time(&self) -> Result<u64, BitmexHttpError> {
        let response: BitmexApiInfo = self
            .send_request::<_, ()>(Method::GET, "", None, None, false)
            .await?;
        Ok(response.timestamp)
    }

    /// Get the instrument definition for the specified symbol.
    ///
    /// BitMEX responds to `/instrument?symbol=...` with an array, even when
    /// a single symbol is requested. This helper returns the first element of
    /// that array and yields `Ok(None)` when the venue returns an empty list
    /// (e.g. unknown symbol).
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the payload cannot be deserialized.
    pub async fn get_instrument(
        &self,
        symbol: &str,
    ) -> Result<Option<BitmexInstrument>, BitmexHttpError> {
        let path = &format!("/instrument?symbol={symbol}");
        let instruments: Vec<BitmexInstrument> = self
            .send_request::<_, ()>(Method::GET, path, None, None, false)
            .await?;

        Ok(instruments.into_iter().next())
    }

    /// Get user wallet information.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_wallet(&self) -> Result<BitmexWallet, BitmexHttpError> {
        let endpoint = "/user/wallet";
        self.send_request::<_, ()>(Method::GET, endpoint, None, None, true)
            .await
    }

    /// Get user margin information for a specific currency.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_margin(&self, currency: &str) -> Result<BitmexMargin, BitmexHttpError> {
        let path = format!("/user/margin?currency={currency}");
        self.send_request::<_, ()>(Method::GET, &path, None, None, true)
            .await
    }

    /// Get user margin information for all currencies.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_all_margins(&self) -> Result<Vec<BitmexMargin>, BitmexHttpError> {
        self.send_request::<_, ()>(Method::GET, "/user/margin?currency=all", None, None, true)
            .await
    }

    /// Get historical trades.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the API returns an error.
    pub async fn get_trades(
        &self,
        params: GetTradeParams,
    ) -> Result<Vec<BitmexTrade>, BitmexHttpError> {
        self.send_request(Method::GET, "/trade", Some(&params), None, false)
            .await
    }

    /// Get bucketed (aggregated) trade data.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the API returns an error.
    pub async fn get_trade_bucketed(
        &self,
        params: GetTradeBucketedParams,
    ) -> Result<Vec<BitmexTradeBin>, BitmexHttpError> {
        self.send_request(Method::GET, "/trade/bucketed", Some(&params), None, false)
            .await
    }

    /// Get current L2 order book rows.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the API returns an error.
    pub async fn get_order_book_l2(
        &self,
        params: GetOrderBookL2Params,
    ) -> Result<Vec<BitmexOrderBookL2>, BitmexHttpError> {
        self.send_request(Method::GET, "/orderBook/L2", Some(&params), None, false)
            .await
    }

    /// Get historical funding rates.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the API returns an error.
    pub async fn get_funding(
        &self,
        params: GetFundingParams,
    ) -> Result<Vec<BitmexFunding>, BitmexHttpError> {
        self.send_request(Method::GET, "/funding", Some(&params), None, false)
            .await
    }

    /// Get user orders.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_orders(
        &self,
        params: GetOrderParams,
    ) -> Result<Vec<BitmexOrder>, BitmexHttpError> {
        self.send_request(Method::GET, "/order", Some(&params), None, true)
            .await
    }

    /// Place a new order.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, order validation fails, or the API returns an error.
    pub async fn place_order(&self, params: PostOrderParams) -> Result<Value, BitmexHttpError> {
        // BitMEX spec requires form-encoded body for POST /order
        let body = serde_urlencoded::to_string(&params)
            .map_err(|e| {
                BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
            })?
            .into_bytes();
        let path = "/order";
        self.send_request::<_, ()>(Method::POST, path, None, Some(body), true)
            .await
    }

    /// Cancel user orders.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, the order doesn't exist, or the API returns an error.
    pub async fn cancel_orders(&self, params: DeleteOrderParams) -> Result<Value, BitmexHttpError> {
        // BitMEX spec requires form-encoded body for DELETE /order
        let body = serde_urlencoded::to_string(&params)
            .map_err(|e| {
                BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
            })?
            .into_bytes();
        let path = "/order";
        self.send_request::<_, ()>(Method::DELETE, path, None, Some(body), true)
            .await
    }

    /// Amend an existing order.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, the order doesn't exist, or the API returns an error.
    pub async fn amend_order(&self, params: PutOrderParams) -> Result<Value, BitmexHttpError> {
        // BitMEX spec requires form-encoded body for PUT /order
        let body = serde_urlencoded::to_string(&params)
            .map_err(|e| {
                BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
            })?
            .into_bytes();
        let path = "/order";
        self.send_request::<_, ()>(Method::PUT, path, None, Some(body), true)
            .await
    }

    /// Cancel all orders.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    ///
    /// # References
    ///
    /// <https://www.bitmex.com/api/explorer/#!/Order/Order_cancelAll>
    pub async fn cancel_all_orders(
        &self,
        params: DeleteAllOrdersParams,
    ) -> Result<Value, BitmexHttpError> {
        self.send_request(Method::DELETE, "/order/all", Some(&params), None, true)
            .await
    }

    /// Set a dead man's switch (cancel all orders after timeout).
    ///
    /// Calling with `timeout=0` disarms the switch.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    ///
    /// # References
    ///
    /// <https://www.bitmex.com/api/explorer/#!/Order/Order_cancelAllAfter>
    pub async fn cancel_all_after(
        &self,
        params: PostCancelAllAfterParams,
    ) -> Result<Value, BitmexHttpError> {
        let body = serde_urlencoded::to_string(&params)
            .map_err(|e| {
                BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
            })?
            .into_bytes();
        self.send_request::<_, ()>(
            Method::POST,
            "/order/cancelAllAfter",
            None,
            Some(body),
            true,
        )
        .await
    }

    /// Get user executions.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_executions(
        &self,
        params: GetExecutionParams,
    ) -> Result<Vec<BitmexExecution>, BitmexHttpError> {
        let query = serde_urlencoded::to_string(&params).map_err(|e| {
            BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
        })?;
        let path = format!("/execution/tradeHistory?{query}");
        self.send_request::<_, ()>(Method::GET, &path, None, None, true)
            .await
    }

    /// Get user positions.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_positions(
        &self,
        params: GetPositionParams,
    ) -> Result<Vec<BitmexPosition>, BitmexHttpError> {
        self.send_request(Method::GET, "/position", Some(&params), None, true)
            .await
    }

    /// Update position leverage.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn update_position_leverage(
        &self,
        params: PostPositionLeverageParams,
    ) -> Result<BitmexPosition, BitmexHttpError> {
        // BitMEX spec requires form-encoded body for POST endpoints
        let body = serde_urlencoded::to_string(&params)
            .map_err(|e| {
                BitmexHttpError::ValidationError(format!("Failed to serialize parameters: {e}"))
            })?
            .into_bytes();
        let path = "/position/leverage";
        self.send_request::<_, ()>(Method::POST, path, None, Some(body), true)
            .await
    }
}

/// Provides a HTTP client for connecting to the [BitMEX](https://bitmex.com) REST API.
///
/// This is the high-level client that wraps the inner client and provides
/// Nautilus-specific functionality for trading operations.
#[derive(Debug)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.bitmex", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.bitmex")
)]
pub struct BitmexHttpClient {
    pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
    pub(crate) order_type_cache: Arc<DashMap<ClientOrderId, OrderType>>,
    clock: &'static AtomicTime,
    inner: Arc<BitmexRawHttpClient>,
    cache_initialized: AtomicBool,
}

impl Clone for BitmexHttpClient {
    fn clone(&self) -> Self {
        let cache_initialized = AtomicBool::new(false);

        let is_initialized = self.cache_initialized.load(Ordering::Acquire);
        if is_initialized {
            cache_initialized.store(true, Ordering::Release);
        }

        Self {
            inner: self.inner.clone(),
            instruments_cache: self.instruments_cache.clone(),
            order_type_cache: self.order_type_cache.clone(),
            cache_initialized,
            clock: self.clock,
        }
    }
}

impl Default for BitmexHttpClient {
    fn default() -> Self {
        Self::new(
            None,
            None,
            None,
            BitmexEnvironment::Mainnet,
            60,
            3,
            1_000,
            10_000,
            10_000,
            BITMEX_DEFAULT_RATE_LIMIT_PER_SECOND,
            BITMEX_DEFAULT_RATE_LIMIT_PER_MINUTE_UNAUTHENTICATED,
            None,
        )
        .expect("Failed to create default BitmexHttpClient")
    }
}

impl BitmexHttpClient {
    /// Creates a new [`BitmexHttpClient`] instance.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client cannot be created.
    #[expect(clippy::too_many_arguments)]
    pub fn new(
        base_url: Option<String>,
        api_key: Option<String>,
        api_secret: Option<String>,
        environment: BitmexEnvironment,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        recv_window_ms: u64,
        max_requests_per_second: u32,
        max_requests_per_minute: u32,
        proxy_url: Option<String>,
    ) -> Result<Self, BitmexHttpError> {
        // Determine the base URL
        let url = base_url.unwrap_or_else(|| match environment {
            BitmexEnvironment::Testnet => BITMEX_HTTP_TESTNET_URL.to_string(),
            BitmexEnvironment::Mainnet => BITMEX_HTTP_URL.to_string(),
        });

        let (key_var, secret_var) = credential_env_vars(environment);
        let api_key = get_or_env_var_opt(api_key, key_var);
        let api_secret = get_or_env_var_opt(api_secret, secret_var);

        let inner = match (api_key, api_secret) {
            (Some(key), Some(secret)) => BitmexRawHttpClient::with_credentials(
                key,
                secret,
                url,
                timeout_secs,
                max_retries,
                retry_delay_ms,
                retry_delay_max_ms,
                recv_window_ms,
                max_requests_per_second,
                max_requests_per_minute,
                proxy_url,
            )?,
            (Some(_), None) | (None, Some(_)) => {
                return Err(BitmexHttpError::ValidationError(
                    "Both api_key and api_secret must be provided, or neither".to_string(),
                ));
            }
            (None, None) => BitmexRawHttpClient::new(
                Some(url),
                timeout_secs,
                max_retries,
                retry_delay_ms,
                retry_delay_max_ms,
                recv_window_ms,
                max_requests_per_second,
                max_requests_per_minute,
                proxy_url,
            )?,
        };

        Ok(Self {
            inner: Arc::new(inner),
            instruments_cache: Arc::new(AtomicMap::new()),
            order_type_cache: Arc::new(DashMap::new()),
            cache_initialized: AtomicBool::new(false),
            clock: get_atomic_clock_realtime(),
        })
    }

    /// Creates a new [`BitmexHttpClient`] instance using environment variables and
    /// the default BitMEX HTTP base URL.
    ///
    /// # Errors
    ///
    /// Returns an error if required environment variables are not set or invalid.
    pub fn from_env() -> anyhow::Result<Self> {
        Self::with_credentials(
            None, None, None, 60, 3, 1_000, 10_000, 10_000, 10, 120, None,
        )
        .map_err(|e| anyhow::anyhow!("Failed to create HTTP client: {e}"))
    }

    /// Creates a new [`BitmexHttpClient`] configured with credentials
    /// for authenticated requests.
    ///
    /// If `api_key` or `api_secret` are `None`, they will be sourced from the
    /// `BITMEX_API_KEY` and `BITMEX_API_SECRET` environment variables.
    ///
    /// # Errors
    ///
    /// Returns an error if one credential is provided without the other.
    #[expect(clippy::too_many_arguments)]
    pub fn with_credentials(
        api_key: Option<String>,
        api_secret: Option<String>,
        base_url: Option<String>,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        recv_window_ms: u64,
        max_requests_per_second: u32,
        max_requests_per_minute: u32,
        proxy_url: Option<String>,
    ) -> anyhow::Result<Self> {
        // Determine environment from URL to select correct environment variables
        let environment = if base_url.as_ref().is_some_and(|url| url.contains("testnet")) {
            BitmexEnvironment::Testnet
        } else {
            BitmexEnvironment::Mainnet
        };

        let (key_var, secret_var) = credential_env_vars(environment);

        let api_key = get_or_env_var_opt(api_key, key_var);
        let api_secret = get_or_env_var_opt(api_secret, secret_var);

        // If we're trying to create an authenticated client, we need both key and secret
        if api_key.is_some() && api_secret.is_none() {
            anyhow::bail!("{secret_var} is required when {key_var} is provided");
        }

        if api_key.is_none() && api_secret.is_some() {
            anyhow::bail!("{key_var} is required when {secret_var} is provided");
        }

        Self::new(
            base_url,
            api_key,
            api_secret,
            environment,
            timeout_secs,
            max_retries,
            retry_delay_ms,
            retry_delay_max_ms,
            recv_window_ms,
            max_requests_per_second,
            max_requests_per_minute,
            proxy_url,
        )
        .map_err(|e| anyhow::anyhow!("Failed to create HTTP client: {e}"))
    }

    /// Returns the base url being used by the client.
    #[must_use]
    pub fn base_url(&self) -> &str {
        self.inner.base_url.as_str()
    }

    /// Returns the public API key being used by the client.
    #[must_use]
    pub fn api_key(&self) -> Option<&str> {
        self.inner.credential.as_ref().map(|c| c.api_key())
    }

    /// Returns a masked version of the API key for logging purposes.
    #[must_use]
    pub fn api_key_masked(&self) -> Option<String> {
        self.inner.credential.as_ref().map(|c| c.api_key_masked())
    }

    /// Requests the current server time from BitMEX.
    ///
    /// Returns the BitMEX system time as a Unix timestamp in milliseconds.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or if the response cannot be parsed.
    pub async fn get_server_time(&self) -> Result<u64, BitmexHttpError> {
        self.inner.get_server_time().await
    }

    /// Sets the dead man's switch (cancel all orders after timeout).
    ///
    /// Calling with `timeout_ms=0` disarms the switch.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails.
    pub async fn cancel_all_after(&self, timeout_ms: u64) -> anyhow::Result<()> {
        let params = PostCancelAllAfterParams {
            timeout: timeout_ms,
        };
        self.inner.cancel_all_after(params).await?;
        Ok(())
    }

    /// Generates a timestamp for initialization.
    fn generate_ts_init(&self) -> UnixNanos {
        self.clock.get_time_ns()
    }

    /// Check if the order has a contingency type that requires linking.
    fn is_contingent_order(contingency_type: ContingencyType) -> bool {
        matches!(
            contingency_type,
            ContingencyType::Oco | ContingencyType::Oto | ContingencyType::Ouo
        )
    }

    /// Check if the order is a parent in contingency relationships.
    fn is_parent_contingency(contingency_type: ContingencyType) -> bool {
        matches!(
            contingency_type,
            ContingencyType::Oco | ContingencyType::Oto
        )
    }

    /// Populate missing `linked_order_ids` for contingency orders by grouping on `order_list_id`.
    fn populate_linked_order_ids(reports: &mut [OrderStatusReport]) {
        let mut order_list_groups: HashMap<OrderListId, Vec<ClientOrderId>> = HashMap::new();
        let mut order_list_parents: HashMap<OrderListId, ClientOrderId> = HashMap::new();
        let mut prefix_groups: HashMap<String, Vec<ClientOrderId>> = HashMap::new();
        let mut prefix_parents: HashMap<String, ClientOrderId> = HashMap::new();

        for report in reports.iter() {
            let Some(client_order_id) = report.client_order_id else {
                continue;
            };

            if let Some(order_list_id) = report.order_list_id {
                order_list_groups
                    .entry(order_list_id)
                    .or_default()
                    .push(client_order_id);

                if Self::is_parent_contingency(report.contingency_type) {
                    order_list_parents
                        .entry(order_list_id)
                        .or_insert(client_order_id);
                }
            }

            if let Some((base, _)) = client_order_id.as_str().rsplit_once('-')
                && Self::is_contingent_order(report.contingency_type)
            {
                prefix_groups
                    .entry(base.to_owned())
                    .or_default()
                    .push(client_order_id);

                if Self::is_parent_contingency(report.contingency_type) {
                    prefix_parents
                        .entry(base.to_owned())
                        .or_insert(client_order_id);
                }
            }
        }

        for report in reports.iter_mut() {
            let Some(client_order_id) = report.client_order_id else {
                continue;
            };

            if report.linked_order_ids.is_some() {
                continue;
            }

            // Only process contingent orders
            if !Self::is_contingent_order(report.contingency_type) {
                continue;
            }

            if let Some(order_list_id) = report.order_list_id
                && let Some(group) = order_list_groups.get(&order_list_id)
            {
                let mut linked: Vec<ClientOrderId> = group
                    .iter()
                    .copied()
                    .filter(|candidate| candidate != &client_order_id)
                    .collect();

                if !linked.is_empty() {
                    if let Some(parent_id) = order_list_parents.get(&order_list_id) {
                        if client_order_id == *parent_id {
                            report.parent_order_id = None;
                        } else {
                            linked.sort_by_key(|candidate| i32::from(candidate != parent_id));
                            report.parent_order_id = Some(*parent_id);
                        }
                    } else {
                        report.parent_order_id = None;
                    }

                    log::trace!(
                        "BitMEX linked ids sourced from order list id: client_order_id={:?}, order_list_id={:?}, contingency_type={:?}, linked_order_ids={:?}",
                        client_order_id,
                        order_list_id,
                        report.contingency_type,
                        linked,
                    );
                    report.linked_order_ids = Some(linked);
                    continue;
                }

                log::trace!(
                    "BitMEX order list id group had no peers: client_order_id={:?}, order_list_id={:?}, contingency_type={:?}, order_list_group={:?}",
                    client_order_id,
                    order_list_id,
                    report.contingency_type,
                    group,
                );
                report.parent_order_id = None;
            } else if report.order_list_id.is_none() {
                report.parent_order_id = None;
            }

            if let Some((base, _)) = client_order_id.as_str().rsplit_once('-')
                && let Some(group) = prefix_groups.get(base)
            {
                let mut linked: Vec<ClientOrderId> = group
                    .iter()
                    .copied()
                    .filter(|candidate| candidate != &client_order_id)
                    .collect();

                if !linked.is_empty() {
                    if let Some(parent_id) = prefix_parents.get(base) {
                        if client_order_id == *parent_id {
                            report.parent_order_id = None;
                        } else {
                            linked.sort_by_key(|candidate| i32::from(candidate != parent_id));
                            report.parent_order_id = Some(*parent_id);
                        }
                    } else {
                        report.parent_order_id = None;
                    }

                    log::trace!(
                        "BitMEX linked ids constructed from client order id prefix: client_order_id={:?}, contingency_type={:?}, base={}, linked_order_ids={:?}",
                        client_order_id,
                        report.contingency_type,
                        base,
                        linked,
                    );
                    report.linked_order_ids = Some(linked);
                    continue;
                }

                log::trace!(
                    "BitMEX client order id prefix group had no peers: client_order_id={:?}, contingency_type={:?}, base={}, prefix_group={:?}",
                    client_order_id,
                    report.contingency_type,
                    base,
                    group,
                );
                report.parent_order_id = None;
            } else if client_order_id.as_str().contains('-') {
                report.parent_order_id = None;
            }

            if report.contingency_type == ContingencyType::Oto {
                log::debug!(
                    "BitMEX OTO order has no linked venue peers; reconciling as standalone: client_order_id={:?}, order_list_id={:?}",
                    report.client_order_id,
                    report.order_list_id,
                );
                report.contingency_type = ContingencyType::NoContingency;
                report.parent_order_id = None;
            } else if Self::is_contingent_order(report.contingency_type) {
                log::warn!(
                    "BitMEX order status report missing linked ids after grouping: client_order_id={:?}, order_list_id={:?}, contingency_type={:?}",
                    report.client_order_id,
                    report.order_list_id,
                    report.contingency_type,
                );
                report.contingency_type = ContingencyType::NoContingency;
                report.parent_order_id = None;
            }

            report.linked_order_ids = None;
        }
    }

    /// Cancel all pending HTTP requests.
    pub fn cancel_all_requests(&self) {
        self.inner.cancel_all_requests();
    }

    /// Replace the cancellation token so new requests can proceed.
    pub fn reset_cancellation_token(&self) {
        self.inner.reset_cancellation_token();
    }

    /// Get a clone of the cancellation token for this client.
    pub fn cancellation_token(&self) -> CancellationToken {
        self.inner.cancellation_token()
    }

    /// Caches a single instrument.
    ///
    /// Any existing instrument with the same symbol will be replaced.
    pub fn cache_instrument(&self, instrument: InstrumentAny) {
        self.instruments_cache
            .insert(instrument.raw_symbol().inner(), instrument);
        self.cache_initialized.store(true, Ordering::Release);
    }

    /// Caches multiple instruments.
    ///
    /// Any existing instruments with the same symbols will be replaced.
    pub fn cache_instruments(&self, instruments: &[InstrumentAny]) {
        self.instruments_cache.rcu(|m| {
            for inst in instruments {
                m.insert(inst.raw_symbol().inner(), inst.clone());
            }
        });
        self.cache_initialized.store(true, Ordering::Release);
    }

    /// Gets an instrument from the cache by symbol.
    pub fn get_instrument(&self, symbol: &Ustr) -> Option<InstrumentAny> {
        self.instruments_cache.get_cloned(symbol)
    }

    /// Request a single instrument and parse it into a Nautilus type.
    ///
    /// # Errors
    ///
    /// Returns `Ok(Some(..))` when the venue returns a definition that parses
    /// successfully, `Ok(None)` when the instrument is unknown, unsupported, or the payload
    /// cannot be converted into a Nautilus `Instrument`.
    pub async fn request_instrument(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Option<InstrumentAny>> {
        let response = self
            .inner
            .get_instrument(instrument_id.symbol.as_str())
            .await?;

        let instrument = match response {
            Some(instrument) => instrument,
            None => return Ok(None),
        };

        let ts_init = self.generate_ts_init();

        match parse_instrument_any(&instrument, ts_init) {
            InstrumentParseResult::Ok(inst) => Ok(Some(*inst)),
            InstrumentParseResult::Unsupported {
                symbol,
                instrument_type,
            } => {
                log::debug!(
                    "Instrument {symbol} has unsupported type {instrument_type:?}, returning None"
                );
                Ok(None)
            }
            InstrumentParseResult::Inactive { symbol, state } => {
                log::debug!("Instrument {symbol} is inactive (state={state}), returning None");
                Ok(None)
            }
            InstrumentParseResult::Failed {
                symbol,
                instrument_type,
                error,
            } => {
                log::error!(
                    "Failed to parse instrument {symbol} (type={instrument_type:?}): {error}"
                );
                Ok(None)
            }
        }
    }

    /// Request all available instruments and parse them into Nautilus types.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    pub async fn request_instruments(
        &self,
        active_only: bool,
    ) -> anyhow::Result<Vec<InstrumentAny>> {
        let instruments = self.inner.get_instruments(active_only).await?;
        let ts_init = self.generate_ts_init();

        let mut parsed_instruments = Vec::new();
        let mut skipped_count = 0;
        let mut inactive_count = 0;
        let mut failed_count = 0;
        let total_count = instruments.len();

        for inst in instruments {
            match parse_instrument_any(&inst, ts_init) {
                InstrumentParseResult::Ok(instrument_any) => {
                    parsed_instruments.push(*instrument_any);
                }
                InstrumentParseResult::Unsupported {
                    symbol,
                    instrument_type,
                } => {
                    skipped_count += 1;
                    log::debug!(
                        "Skipping unsupported instrument type: symbol={symbol}, type={instrument_type:?}"
                    );
                }
                InstrumentParseResult::Inactive { symbol, state } => {
                    inactive_count += 1;
                    log::debug!("Skipping inactive instrument: symbol={symbol}, state={state}");
                }
                InstrumentParseResult::Failed {
                    symbol,
                    instrument_type,
                    error,
                } => {
                    failed_count += 1;
                    log::error!(
                        "Failed to parse instrument: symbol={symbol}, type={instrument_type:?}, error={error}"
                    );
                }
            }
        }

        if skipped_count > 0 {
            log::debug!(
                "Skipped {skipped_count} unsupported instrument type(s) out of {total_count} total"
            );
        }

        if inactive_count > 0 {
            log::debug!(
                "Skipped {inactive_count} inactive instrument(s) out of {total_count} total"
            );
        }

        if failed_count > 0 {
            log::error!(
                "Instrument parse failures: {failed_count} failed out of {total_count} total ({} successfully parsed)",
                parsed_instruments.len()
            );
        }

        Ok(parsed_instruments)
    }

    /// Get user wallet information.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_wallet(&self) -> Result<BitmexWallet, BitmexHttpError> {
        let inner = self.inner.clone();
        inner.get_wallet().await
    }

    /// Get user orders.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_orders(
        &self,
        params: GetOrderParams,
    ) -> Result<Vec<BitmexOrder>, BitmexHttpError> {
        let inner = self.inner.clone();
        inner.get_orders(params).await
    }

    /// Get instrument from the instruments cache (if found).
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument is not found in the cache.
    fn instrument_from_cache(&self, symbol: Ustr) -> anyhow::Result<InstrumentAny> {
        self.get_instrument(&symbol).ok_or_else(|| {
            anyhow::anyhow!(
                "Instrument {symbol} not found in cache, ensure instruments loaded first"
            )
        })
    }

    /// Returns the cached price precision for the given symbol.
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument was never cached (for example, if
    /// instruments were not loaded prior to use).
    pub fn get_price_precision(&self, symbol: Ustr) -> anyhow::Result<u8> {
        self.instrument_from_cache(symbol)
            .map(|instrument| instrument.price_precision())
    }

    /// Get user margin information for a specific currency.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_margin(&self, currency: &str) -> anyhow::Result<BitmexMargin> {
        self.inner
            .get_margin(currency)
            .await
            .map_err(|e| anyhow::anyhow!(e))
    }

    /// Get user margin information for all currencies.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, or the API returns an error.
    pub async fn get_all_margins(&self) -> anyhow::Result<Vec<BitmexMargin>> {
        self.inner
            .get_all_margins()
            .await
            .map_err(|e| anyhow::anyhow!(e))
    }

    /// Request account state for the authenticated BitMEX account.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or no account state is returned.
    pub async fn request_account_state(
        &self,
        fallback_account_id: AccountId,
    ) -> anyhow::Result<AccountState> {
        let margins = self
            .inner
            .get_all_margins()
            .await
            .map_err(|e| anyhow::anyhow!(e))?;
        let account_id = account_id_from_margins(&margins)?.unwrap_or(fallback_account_id);

        let ts_init =
            UnixNanos::from(chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default() as u64);

        let mut balances = Vec::with_capacity(margins.len());
        let mut margins_vec = Vec::new();
        let mut latest_timestamp: Option<chrono::DateTime<chrono::Utc>> = None;

        for margin in margins {
            if let Some(ts) = margin.timestamp {
                latest_timestamp = Some(latest_timestamp.map_or(ts, |prev| prev.max(ts)));
            }

            let margin_msg = BitmexMarginMsg {
                account: margin.account,
                currency: margin.currency,
                risk_limit: margin.risk_limit,
                amount: margin.amount,
                prev_realised_pnl: margin.prev_realised_pnl,
                gross_comm: margin.gross_comm,
                gross_open_cost: margin.gross_open_cost,
                gross_open_premium: margin.gross_open_premium,
                gross_exec_cost: margin.gross_exec_cost,
                gross_mark_value: margin.gross_mark_value,
                risk_value: margin.risk_value,
                init_margin: margin.init_margin,
                maint_margin: margin.maint_margin,
                target_excess_margin: margin.target_excess_margin,
                realised_pnl: margin.realised_pnl,
                unrealised_pnl: margin.unrealised_pnl,
                wallet_balance: margin.wallet_balance,
                margin_balance: margin.margin_balance,
                margin_leverage: margin.margin_leverage,
                margin_used_pcnt: margin.margin_used_pcnt,
                excess_margin: margin.excess_margin,
                available_margin: margin.available_margin,
                withdrawable_margin: margin.withdrawable_margin,
                maker_fee_discount: None,
                taker_fee_discount: None,
                timestamp: margin.timestamp.unwrap_or_else(chrono::Utc::now),
                foreign_margin_balance: None,
                foreign_requirement: None,
            };

            let balance = parse_account_balance(&margin_msg);

            let divisor = bitmex_currency_divisor(margin_msg.currency.as_str());
            let initial_dec = Decimal::from(margin_msg.init_margin.unwrap_or(0).max(0)) / divisor;
            let maintenance_dec =
                Decimal::from(margin_msg.maint_margin.unwrap_or(0).max(0)) / divisor;

            if !initial_dec.is_zero() || !maintenance_dec.is_zero() {
                let currency = balance.total.currency;
                // BitMEX reports cross-margin aggregates per collateral currency.
                margins_vec.push(MarginBalance::new(
                    Money::from_decimal(initial_dec, currency)
                        .unwrap_or_else(|_| Money::zero(currency)),
                    Money::from_decimal(maintenance_dec, currency)
                        .unwrap_or_else(|_| Money::zero(currency)),
                    None,
                ));
            }

            balances.push(balance);
        }

        if balances.is_empty() {
            anyhow::bail!("No margin data returned from BitMEX");
        }

        let account_type = AccountType::Margin;
        let is_reported = true;
        let event_id = UUID4::new();

        // Use server timestamp if available, otherwise fall back to local time
        let ts_event = latest_timestamp.map_or(ts_init, |ts| {
            UnixNanos::from(ts.timestamp_nanos_opt().unwrap_or_default() as u64)
        });

        Ok(AccountState::new(
            account_id,
            account_type,
            balances,
            margins_vec,
            is_reported,
            event_id,
            ts_event,
            ts_init,
            None,
        ))
    }

    /// Submit a new order.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, order validation fails,
    /// the order is rejected, or the API returns an error.
    #[expect(clippy::too_many_arguments)]
    pub async fn submit_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: TimeInForce,
        price: Option<Price>,
        trigger_price: Option<Price>,
        trigger_type: Option<TriggerType>,
        trailing_offset: Option<f64>,
        trailing_offset_type: Option<TrailingOffsetType>,
        display_qty: Option<Quantity>,
        post_only: bool,
        reduce_only: bool,
        order_list_id: Option<OrderListId>,
        contingency_type: Option<ContingencyType>,
        peg_price_type: Option<BitmexPegPriceType>,
        peg_offset_value: Option<f64>,
    ) -> anyhow::Result<OrderStatusReport> {
        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;

        let mut params = super::query::PostOrderParamsBuilder::default();
        params.text(NAUTILUS_TRADER);
        params.symbol(instrument_id.symbol.as_str());
        params.cl_ord_id(client_order_id.as_str());

        if order_side == OrderSide::NoOrderSide {
            anyhow::bail!("Order side must be Buy or Sell");
        }
        let side = BitmexSide::from(order_side.as_specified());
        params.side(side);

        let ord_type = BitmexOrderType::try_from_order_type(order_type)?;
        params.ord_type(ord_type);

        params.order_qty(quantity_to_u32(&quantity, &instrument));

        let tif = BitmexTimeInForce::try_from_time_in_force(time_in_force)?;
        params.time_in_force(tif);

        if let Some(price) = price {
            params.price(price.as_f64());
        }

        if let Some(trigger_price) = trigger_price {
            params.stop_px(trigger_price.as_f64());
        }

        if let Some(display_qty) = display_qty {
            params.display_qty(quantity_to_u32(&display_qty, &instrument));
        }

        if let Some(order_list_id) = order_list_id {
            params.cl_ord_link_id(order_list_id.as_str());
        }

        let is_trailing_stop = matches!(
            order_type,
            OrderType::TrailingStopMarket | OrderType::TrailingStopLimit
        );

        if is_trailing_stop && let Some(offset) = trailing_offset {
            if let Some(offset_type) = trailing_offset_type
                && offset_type != TrailingOffsetType::Price
            {
                anyhow::bail!(
                    "BitMEX only supports PRICE trailing offset type, was {offset_type:?}"
                );
            }

            params.peg_price_type(BitmexPegPriceType::TrailingStopPeg);

            // BitMEX requires negative offset for stop-sell orders
            let signed_offset = match order_side {
                OrderSide::Sell => -offset.abs(),
                OrderSide::Buy => offset.abs(),
                _ => offset,
            };
            params.peg_offset_value(signed_offset);
        }

        // Pegged orders (BBO) via params override
        if peg_price_type.is_none() && peg_offset_value.is_some() {
            anyhow::bail!("`peg_offset_value` requires `peg_price_type`");
        }

        if let Some(peg_type) = peg_price_type {
            if order_type != OrderType::Limit {
                anyhow::bail!(
                    "Pegged orders only supported for LIMIT order type, was {order_type:?}"
                );
            }
            params.ord_type(BitmexOrderType::Pegged);
            params.peg_price_type(peg_type);

            if let Some(offset) = peg_offset_value {
                params.peg_offset_value(offset);
            }
        }

        let mut exec_inst = Vec::new();

        if post_only {
            exec_inst.push(BitmexExecInstruction::ParticipateDoNotInitiate);
        }

        if reduce_only {
            exec_inst.push(BitmexExecInstruction::ReduceOnly);
        }

        // For trailing stops, trigger_type specifies which price to track (Mark, Last, Index)
        if (trigger_price.is_some() || is_trailing_stop)
            && let Some(trigger_type) = trigger_type
        {
            match trigger_type {
                TriggerType::LastPrice => exec_inst.push(BitmexExecInstruction::LastPrice),
                TriggerType::MarkPrice => exec_inst.push(BitmexExecInstruction::MarkPrice),
                TriggerType::IndexPrice => exec_inst.push(BitmexExecInstruction::IndexPrice),
                _ => {} // Use BitMEX default (LastPrice) for other trigger types
            }
        }

        if !exec_inst.is_empty() {
            params.exec_inst(exec_inst);
        }

        if let Some(contingency_type) = contingency_type {
            let bitmex_contingency = BitmexContingencyType::try_from(contingency_type)?;
            params.contingency_type(bitmex_contingency);
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.place_order(params).await?;

        let order: BitmexOrder = serde_json::from_value(response)?;

        if order.ord_status == Some(BitmexOrderStatus::Rejected) {
            let reason = order
                .ord_rej_reason
                .map_or_else(|| "No reason provided".to_string(), |r| r.to_string());
            anyhow::bail!("Order rejected: {reason}");
        }

        // Cache order type for future lookups (e.g., cancel responses missing ord_type)
        self.order_type_cache.insert(client_order_id, order_type);

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        parse_order_status_report(&order, &instrument, &self.order_type_cache, ts_init)
    }

    /// Cancel an order.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The order doesn't exist.
    /// - The API returns an error.
    pub async fn cancel_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> anyhow::Result<OrderStatusReport> {
        let mut params = super::query::DeleteOrderParamsBuilder::default();
        params.text(NAUTILUS_TRADER);

        if let Some(venue_order_id) = venue_order_id {
            params.order_id(vec![venue_order_id.as_str().to_string()]);
        } else if let Some(client_order_id) = client_order_id {
            params.cl_ord_id(vec![client_order_id.as_str().to_string()]);
        } else {
            anyhow::bail!("Either client_order_id or venue_order_id must be provided");
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.cancel_orders(params).await?;

        let orders: Vec<BitmexOrder> = serde_json::from_value(response)?;
        let order = orders
            .into_iter()
            .next()
            .ok_or_else(|| anyhow::anyhow!("No order returned in cancel response"))?;

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        parse_order_status_report(&order, &instrument, &self.order_type_cache, ts_init)
    }

    /// Cancel multiple orders.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The order doesn't exist.
    /// - The API returns an error.
    pub async fn cancel_orders(
        &self,
        instrument_id: InstrumentId,
        client_order_ids: Option<Vec<ClientOrderId>>,
        venue_order_ids: Option<Vec<VenueOrderId>>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let mut params = super::query::DeleteOrderParamsBuilder::default();
        params.text(NAUTILUS_TRADER);

        // BitMEX API requires either client order IDs or venue order IDs, not both
        // Prioritize venue order IDs if both are provided
        if let Some(venue_order_ids) = venue_order_ids {
            if venue_order_ids.is_empty() {
                anyhow::bail!("venue_order_ids cannot be empty");
            }
            params.order_id(
                venue_order_ids
                    .iter()
                    .map(|id| id.to_string())
                    .collect::<Vec<_>>(),
            );
        } else if let Some(client_order_ids) = client_order_ids {
            if client_order_ids.is_empty() {
                anyhow::bail!("client_order_ids cannot be empty");
            }
            params.cl_ord_id(
                client_order_ids
                    .iter()
                    .map(|id| id.to_string())
                    .collect::<Vec<_>>(),
            );
        } else {
            anyhow::bail!("Either client_order_ids or venue_order_ids must be provided");
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.cancel_orders(params).await?;

        let orders: Vec<BitmexOrder> = serde_json::from_value(response)?;

        let ts_init = self.generate_ts_init();
        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;

        let mut reports = Vec::new();

        for order in orders {
            reports.push(parse_order_status_report(
                &order,
                &instrument,
                &self.order_type_cache,
                ts_init,
            )?);
        }

        Self::populate_linked_order_ids(&mut reports);

        Ok(reports)
    }

    /// Cancel all orders for an instrument and optionally an order side.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The order doesn't exist.
    /// - The API returns an error.
    pub async fn cancel_all_orders(
        &self,
        instrument_id: InstrumentId,
        order_side: Option<OrderSide>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let mut params = DeleteAllOrdersParamsBuilder::default();
        params.text(NAUTILUS_TRADER);
        params.symbol(instrument_id.symbol.as_str());

        if let Some(side) = order_side {
            if side == OrderSide::NoOrderSide {
                log::debug!("Ignoring NoOrderSide filter for cancel_all_orders on {instrument_id}",);
            } else {
                let side = BitmexSide::from(side.as_specified());
                params.filter(serde_json::json!({
                    "side": side
                }));
            }
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.cancel_all_orders(params).await?;

        let orders: Vec<BitmexOrder> = serde_json::from_value(response)?;

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        let mut reports = Vec::new();

        for order in orders {
            reports.push(parse_order_status_report(
                &order,
                &instrument,
                &self.order_type_cache,
                ts_init,
            )?);
        }

        Self::populate_linked_order_ids(&mut reports);

        Ok(reports)
    }

    /// Modify an existing order.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The order doesn't exist.
    /// - The order is already closed.
    /// - The API returns an error.
    pub async fn modify_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
        quantity: Option<Quantity>,
        price: Option<Price>,
        trigger_price: Option<Price>,
    ) -> anyhow::Result<OrderStatusReport> {
        let mut params = PutOrderParamsBuilder::default();
        params.text(NAUTILUS_TRADER);

        // Set order ID - prefer venue_order_id if available
        if let Some(venue_order_id) = venue_order_id {
            params.order_id(venue_order_id.as_str());
        } else if let Some(client_order_id) = client_order_id {
            params.orig_cl_ord_id(client_order_id.as_str());
        } else {
            anyhow::bail!("Either client_order_id or venue_order_id must be provided");
        }

        if let Some(quantity) = quantity {
            let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
            params.order_qty(quantity_to_u32(&quantity, &instrument));
        }

        if let Some(price) = price {
            params.price(price.as_f64());
        }

        if let Some(trigger_price) = trigger_price {
            params.stop_px(trigger_price.as_f64());
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.amend_order(params).await?;

        let order: BitmexOrder = serde_json::from_value(response)?;

        if order.ord_status == Some(BitmexOrderStatus::Rejected) {
            let reason = order
                .ord_rej_reason
                .map_or_else(|| "No reason provided".to_string(), |r| r.to_string());
            anyhow::bail!("Order modification rejected: {reason}");
        }

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        parse_order_status_report(&order, &instrument, &self.order_type_cache, ts_init)
    }

    /// Query a single order by client order ID or venue order ID.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The API returns an error.
    pub async fn query_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let mut params = GetOrderParamsBuilder::default();

        let filter_json = if let Some(client_order_id) = client_order_id {
            serde_json::json!({
                "clOrdID": client_order_id.to_string()
            })
        } else if let Some(venue_order_id) = venue_order_id {
            serde_json::json!({
                "orderID": venue_order_id.to_string()
            })
        } else {
            anyhow::bail!("Either client_order_id or venue_order_id must be provided");
        };

        params.filter(filter_json);
        params.count(1); // Only need one order

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_orders(params).await?;

        if response.is_empty() {
            return Ok(None);
        }

        let order = &response[0];

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        let report =
            parse_order_status_report(order, &instrument, &self.order_type_cache, ts_init)?;

        Ok(Some(report))
    }

    /// Request a single order status report.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The API returns an error.
    pub async fn request_order_status_report(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> anyhow::Result<OrderStatusReport> {
        if venue_order_id.is_none() && client_order_id.is_none() {
            anyhow::bail!("Either venue_order_id or client_order_id must be provided");
        }

        let mut params = GetOrderParamsBuilder::default();
        params.symbol(instrument_id.symbol.as_str());

        if let Some(venue_order_id) = venue_order_id {
            params.filter(serde_json::json!({
                "orderID": venue_order_id.as_str()
            }));
        } else if let Some(client_order_id) = client_order_id {
            params.filter(serde_json::json!({
                "clOrdID": client_order_id.as_str()
            }));
        }

        params.count(1i32);
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_orders(params).await?;

        let order = response
            .into_iter()
            .next()
            .ok_or_else(|| anyhow::anyhow!("Order not found"))?;

        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        parse_order_status_report(&order, &instrument, &self.order_type_cache, ts_init)
    }

    /// Request multiple order status reports.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Credentials are missing.
    /// - The request fails.
    /// - The API returns an error.
    pub async fn request_order_status_reports(
        &self,
        instrument_id: Option<InstrumentId>,
        open_only: bool,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        if let (Some(start), Some(end)) = (start, end) {
            anyhow::ensure!(
                start < end,
                "Invalid time range: start={start:?} end={end:?}",
            );
        }

        let mut params = GetOrderParamsBuilder::default();

        if let Some(instrument_id) = &instrument_id {
            params.symbol(instrument_id.symbol.as_str());
        }

        if open_only {
            params.filter(serde_json::json!({
                "open": true
            }));
        }

        if let Some(start) = start {
            params.start_time(start);
        }

        if let Some(end) = end {
            params.end_time(end);
        }

        if let Some(limit) = limit {
            params.count(limit as i32);
        } else {
            params.count(500); // Default count to avoid empty query
        }

        params.reverse(true); // Get newest orders first

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_orders(params).await?;

        let ts_init = self.generate_ts_init();

        let mut reports = Vec::new();

        for order in response {
            if let Some(start) = start {
                match order.timestamp {
                    Some(timestamp) if timestamp < start => continue,
                    Some(_) => {}
                    None => {
                        log::debug!("Skipping order report without timestamp for bounded query");
                        continue;
                    }
                }
            }

            if let Some(end) = end {
                match order.timestamp {
                    Some(timestamp) if timestamp > end => continue,
                    Some(_) => {}
                    None => {
                        log::debug!("Skipping order report without timestamp for bounded query");
                        continue;
                    }
                }
            }

            // Skip orders without symbol (can happen with query responses)
            let Some(symbol) = order.symbol else {
                log::warn!("Order response missing symbol, skipping");
                continue;
            };

            let Ok(instrument) = self.instrument_from_cache(symbol) else {
                log::debug!("Skipping order report for instrument not in cache: symbol={symbol}");
                continue;
            };

            match parse_order_status_report(&order, &instrument, &self.order_type_cache, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => log::error!("Failed to parse order status report: {e}"),
            }
        }

        Self::populate_linked_order_ids(&mut reports);

        Ok(reports)
    }

    /// Request trades for the given instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    pub async fn request_trades(
        &self,
        instrument_id: InstrumentId,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<TradeTick>> {
        let mut params = GetTradeParamsBuilder::default();
        params.symbol(instrument_id.symbol.as_str());

        if let Some(start) = start {
            params.start_time(start);
        }

        if let Some(end) = end {
            params.end_time(end);
        }

        if let (Some(start), Some(end)) = (start, end) {
            anyhow::ensure!(
                start < end,
                "Invalid time range: start={start:?} end={end:?}",
            );
        }

        if let Some(limit) = limit {
            let clamped_limit = limit.min(1000);
            if limit > 1000 {
                log::warn!(
                    "BitMEX trade request limit exceeds venue maximum; clamping: limit={limit}, clamped_limit={clamped_limit}",
                );
            }
            params.count(i32::try_from(clamped_limit).unwrap_or(1000));
        }
        params.reverse(false);
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_trades(params).await?;

        let ts_init = self.generate_ts_init();

        let mut parsed_trades = Vec::new();

        for trade in response {
            if let Some(start) = start
                && trade.timestamp < start
            {
                continue;
            }

            if let Some(end) = end
                && trade.timestamp > end
            {
                continue;
            }

            let Some(instrument) = self.get_instrument(&trade.symbol) else {
                log::error!(
                    "Instrument {} not found in cache, skipping trade",
                    trade.symbol
                );
                continue;
            };

            match parse_trade(&trade, &instrument, ts_init) {
                Ok(trade) => parsed_trades.push(trade),
                Err(e) => log::error!("Failed to parse trade: {e}"),
            }
        }

        Ok(parsed_trades)
    }

    /// Request bars for the given bar type.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails, parsing fails, or the bar specification is
    /// unsupported by BitMEX.
    pub async fn request_bars(
        &self,
        mut bar_type: BarType,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
        partial: bool,
    ) -> anyhow::Result<Vec<Bar>> {
        bar_type = bar_type.standard();

        anyhow::ensure!(
            bar_type.aggregation_source() == AggregationSource::External,
            "Only EXTERNAL aggregation bars are supported"
        );
        anyhow::ensure!(
            bar_type.spec().price_type == PriceType::Last,
            "Only LAST price type bars are supported"
        );

        if let (Some(start), Some(end)) = (start, end) {
            anyhow::ensure!(
                start < end,
                "Invalid time range: start={start:?} end={end:?}"
            );
        }

        let spec = bar_type.spec();
        let bin_size = match (spec.aggregation, spec.step.get()) {
            (BarAggregation::Minute, 1) => "1m",
            (BarAggregation::Minute, 5) => "5m",
            (BarAggregation::Hour, 1) => "1h",
            (BarAggregation::Day, 1) => "1d",
            _ => anyhow::bail!(
                "BitMEX does not support {}-{:?}-{:?} bars",
                spec.step.get(),
                spec.aggregation,
                spec.price_type,
            ),
        };

        let instrument_id = bar_type.instrument_id();
        let instrument = self.instrument_from_cache_by_id(instrument_id)?;

        let mut params = GetTradeBucketedParamsBuilder::default();
        params.symbol(instrument_id.symbol.as_str());
        params.bin_size(bin_size);

        if partial {
            params.partial(true);
        }

        if let Some(start) = start {
            params.start_time(start);
        }

        if let Some(end) = end {
            params.end_time(end);
        }

        if let Some(limit) = limit {
            let clamped_limit = limit.min(1000);
            if limit > 1000 {
                log::warn!(
                    "BitMEX bar request limit exceeds venue maximum; clamping: limit={limit}, clamped_limit={clamped_limit}",
                );
            }
            params.count(i32::try_from(clamped_limit).unwrap_or(1000));
        }
        params.reverse(false);
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_trade_bucketed(params).await?;
        let ts_init = self.generate_ts_init();
        let mut bars = Vec::new();

        for bin in response {
            if let Some(start) = start
                && bin.timestamp < start
            {
                continue;
            }

            if let Some(end) = end
                && bin.timestamp > end
            {
                continue;
            }

            if bin.symbol != instrument_id.symbol.inner() {
                log::warn!(
                    "Skipping trade bin for unexpected symbol: symbol={}, expected={}",
                    bin.symbol,
                    instrument_id.symbol,
                );
                continue;
            }

            match parse_trade_bin(&bin, &instrument, &bar_type, ts_init) {
                Ok(bar) => bars.push(bar),
                Err(e) => log::warn!("Failed to parse trade bin: {e}"),
            }
        }

        Ok(bars)
    }

    /// Request a current L2 order book snapshot.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails, the instrument is not cached, or the book
    /// rows cannot be parsed.
    pub async fn request_book_snapshot(
        &self,
        instrument_id: InstrumentId,
        depth: Option<u32>,
    ) -> anyhow::Result<OrderBook> {
        let instrument = self.instrument_from_cache_by_id(instrument_id)?;
        let mut params = GetOrderBookL2ParamsBuilder::default();
        params.symbol(instrument_id.symbol.as_str());

        if let Some(depth) = depth {
            params.depth(depth);
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;
        let response = self.inner.get_order_book_l2(params).await?;
        let ts_init = self.generate_ts_init();
        let deltas = parse_order_book_l2_snapshot(&response, &instrument, instrument_id, ts_init)?;

        let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);
        book.apply_deltas(&deltas)?;
        Ok(book)
    }

    fn instrument_from_cache_by_id(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<InstrumentAny> {
        self.get_instrument(&instrument_id.symbol.inner())
            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id).into())
    }

    /// Request historical funding rates for the given instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the time range is invalid.
    pub async fn request_funding_rates(
        &self,
        instrument_id: InstrumentId,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<FundingRateUpdate>> {
        if let (Some(start), Some(end)) = (start, end) {
            anyhow::ensure!(
                start < end,
                "Invalid time range: start={start:?} end={end:?}",
            );
        }

        let total_limit = limit.map(|value| value as usize);
        let mut offset = 0_i32;
        let mut rates = Vec::new();

        loop {
            if total_limit.is_some_and(|limit| rates.len() >= limit) {
                break;
            }

            let remaining = total_limit.map_or(BITMEX_MAX_TABLE_COUNT as usize, |limit| {
                limit.saturating_sub(rates.len())
            });
            let page_count = remaining.min(BITMEX_MAX_TABLE_COUNT as usize);

            if page_count == 0 {
                break;
            }

            let mut params = GetFundingParamsBuilder::default();
            params.symbol(instrument_id.symbol.as_str());
            params.count(i32::try_from(page_count).unwrap_or(BITMEX_MAX_TABLE_COUNT as i32));
            params.start(offset);
            params.reverse(false);

            if let Some(start) = start {
                params.start_time(start);
            }

            if let Some(end) = end {
                params.end_time(end);
            }

            let params = params.build().map_err(|e| anyhow::anyhow!(e))?;
            let response = self.inner.get_funding(params).await?;
            let response_len = response.len();

            if response.is_empty() {
                break;
            }

            for raw in response {
                if raw.symbol != instrument_id.symbol.inner() {
                    log::warn!(
                        "Skipping funding rate for unexpected symbol: symbol={}, expected={}",
                        raw.symbol,
                        instrument_id.symbol,
                    );
                    continue;
                }

                if let Some(start) = start
                    && raw.timestamp < start
                {
                    continue;
                }

                if let Some(end) = end
                    && raw.timestamp > end
                {
                    continue;
                }

                let Some(rate) = parse_funding_rate_update(&raw, instrument_id) else {
                    continue;
                };

                rates.push(rate);

                if total_limit.is_some_and(|limit| rates.len() >= limit) {
                    break;
                }
            }

            if response_len < page_count {
                break;
            }

            offset += i32::try_from(response_len).unwrap_or(BITMEX_MAX_TABLE_COUNT as i32);
        }

        Ok(rates)
    }

    /// Request fill reports for the given instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    pub async fn request_fill_reports(
        &self,
        instrument_id: Option<InstrumentId>,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<FillReport>> {
        if let (Some(start), Some(end)) = (start, end) {
            anyhow::ensure!(
                start < end,
                "Invalid time range: start={start:?} end={end:?}",
            );
        }

        let mut params = GetExecutionParamsBuilder::default();

        if let Some(instrument_id) = instrument_id {
            params.symbol(instrument_id.symbol.as_str());
        }

        if let Some(start) = start {
            params.start_time(start);
        }

        if let Some(end) = end {
            params.end_time(end);
        }

        if let Some(limit) = limit {
            params.count(limit as i32);
        } else {
            params.count(500); // Default count
        }
        params.reverse(true); // Get newest fills first

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_executions(params).await?;

        let ts_init = self.generate_ts_init();

        let mut reports = Vec::new();

        for exec in response {
            if let Some(start) = start {
                match exec.transact_time {
                    Some(timestamp) if timestamp < start => continue,
                    Some(_) => {}
                    None => {
                        log::debug!("Skipping fill report without transact_time for bounded query");
                        continue;
                    }
                }
            }

            if let Some(end) = end {
                match exec.transact_time {
                    Some(timestamp) if timestamp > end => continue,
                    Some(_) => {}
                    None => {
                        log::debug!("Skipping fill report without transact_time for bounded query");
                        continue;
                    }
                }
            }

            // Skip executions without symbol (e.g., CancelReject)
            let Some(symbol) = exec.symbol else {
                log::debug!("Skipping execution without symbol: {:?}", exec.exec_type);
                continue;
            };
            let symbol_str = symbol.to_string();

            let instrument = match self.instrument_from_cache(symbol) {
                Ok(instrument) => instrument,
                Err(e) => {
                    log::error!(
                        "Instrument not found in cache for execution parsing: symbol={symbol_str}, {e}"
                    );
                    continue;
                }
            };

            match parse_fill_report(&exec, &instrument, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    // Log at debug level for expected skip cases
                    let error_msg = e.to_string();
                    if error_msg.starts_with("Skipping non-trade execution")
                        || error_msg.starts_with("Skipping execution without order_id")
                    {
                        log::debug!("{e}");
                    } else {
                        log::error!("Failed to parse fill report: {e}");
                    }
                }
            }
        }

        Ok(reports)
    }

    /// Request position reports.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    pub async fn request_position_status_reports(
        &self,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        let params = GetPositionParamsBuilder::default()
            .count(500) // Default count
            .build()
            .map_err(|e| anyhow::anyhow!(e))?;

        let response = self.inner.get_positions(params).await?;

        let ts_init = self.generate_ts_init();

        let mut reports = Vec::new();

        for pos in response {
            let symbol = pos.symbol;
            let instrument = match self.instrument_from_cache(symbol) {
                Ok(instrument) => instrument,
                Err(e) => {
                    log::error!(
                        "Instrument not found in cache for position parsing: symbol={}, {e}",
                        pos.symbol.as_str(),
                    );
                    continue;
                }
            };

            match parse_position_report(&pos, &instrument, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => log::error!("Failed to parse position report: {e}"),
            }
        }

        Ok(reports)
    }

    /// Update position leverage.
    ///
    /// # Errors
    ///
    /// - Credentials are missing.
    /// - The request fails.
    /// - The API returns an error.
    pub async fn update_position_leverage(
        &self,
        symbol: &str,
        leverage: f64,
    ) -> anyhow::Result<PositionStatusReport> {
        let params = PostPositionLeverageParams {
            symbol: symbol.to_string(),
            leverage,
            target_account_id: None,
        };

        let response = self.inner.update_position_leverage(params).await?;

        let instrument = self.instrument_from_cache(Ustr::from(symbol))?;
        let ts_init = self.generate_ts_init();

        parse_position_report(&response, &instrument, ts_init)
    }
}

fn parse_order_book_l2_snapshot(
    rows: &[BitmexOrderBookL2],
    instrument: &InstrumentAny,
    instrument_id: InstrumentId,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderBookDeltas> {
    let price_precision = instrument.price_precision();
    let mut deltas = Vec::with_capacity(rows.len() + 1);
    deltas.push(OrderBookDelta::clear(instrument_id, 0, ts_init, ts_init));

    for row in rows {
        if row.symbol != instrument_id.symbol.inner() {
            log::warn!(
                "Skipping BitMEX order book row for unexpected symbol: symbol={}, expected={}",
                row.symbol,
                instrument_id.symbol,
            );
            continue;
        }

        let Some(price_value) = row.price else {
            log::warn!(
                "Skipping BitMEX order book row without price: symbol={}, id={}",
                row.symbol,
                row.id,
            );
            continue;
        };

        let Some(size_value) = row.size else {
            log::warn!(
                "Skipping BitMEX order book row without size: symbol={}, id={}",
                row.symbol,
                row.id,
            );
            continue;
        };

        let Ok(size) = u64::try_from(size_value) else {
            log::warn!(
                "Skipping BitMEX order book row with negative size: symbol={}, id={}, size={}",
                row.symbol,
                row.id,
                size_value,
            );
            continue;
        };

        let Ok(order_id) = u64::try_from(row.id) else {
            log::warn!(
                "Skipping BitMEX order book row with negative id: symbol={}, id={}",
                row.symbol,
                row.id,
            );
            continue;
        };

        let order = BookOrder::new(
            OrderSide::from(row.side),
            Price::new(price_value, price_precision),
            parse_contracts_quantity(size, instrument),
            order_id,
        );
        let delta = OrderBookDelta::new(
            instrument_id,
            BookAction::Add,
            order,
            RecordFlag::F_SNAPSHOT as u8,
            0,
            ts_init,
            ts_init,
        );
        deltas.push(delta);
    }

    if let Some(last) = deltas.last_mut() {
        last.flags |= RecordFlag::F_LAST as u8;
    }

    OrderBookDeltas::new_checked(instrument_id, deltas)
}

fn parse_funding_rate_update(
    raw: &BitmexFunding,
    instrument_id: InstrumentId,
) -> Option<FundingRateUpdate> {
    let Some(rate) = raw.funding_rate else {
        log::warn!(
            "Skipping BitMEX funding rate without funding_rate: symbol={}, timestamp={}",
            raw.symbol,
            raw.timestamp,
        );
        return None;
    };

    let interval = raw.funding_interval.map(|interval| {
        let minutes = interval.hour() * 60 + interval.minute();
        minutes as u16
    });
    let ts_event = UnixNanos::from(raw.timestamp);

    Some(FundingRateUpdate::new(
        instrument_id,
        rate,
        interval,
        None,
        ts_event,
        ts_event,
    ))
}

fn account_id_from_margins(margins: &[BitmexMargin]) -> anyhow::Result<Option<AccountId>> {
    let Some(first) = margins.first() else {
        return Ok(None);
    };

    let account = first.account;
    if let Some(mismatch) = margins.iter().find(|margin| margin.account != account) {
        anyhow::bail!(
            "BitMEX returned inconsistent margin account IDs: {account} and {}",
            mismatch.account
        );
    }

    Ok(Some(bitmex_account_id(account)))
}

#[cfg(test)]
mod tests {
    use nautilus_core::UUID4;
    use nautilus_model::enums::OrderStatus;
    use rstest::rstest;
    use serde_json::json;

    use super::*;

    fn margin_with_account(account: i64) -> BitmexMargin {
        BitmexMargin {
            account,
            currency: Ustr::from("XBt"),
            risk_limit: None,
            prev_state: None,
            state: None,
            action: None,
            amount: None,
            pending_credit: None,
            pending_debit: None,
            confirmed_debit: None,
            prev_realised_pnl: None,
            prev_unrealised_pnl: None,
            gross_comm: None,
            gross_open_cost: None,
            gross_open_premium: None,
            gross_exec_cost: None,
            gross_mark_value: None,
            risk_value: None,
            taxable_margin: None,
            init_margin: None,
            maint_margin: None,
            session_margin: None,
            target_excess_margin: None,
            var_margin: None,
            realised_pnl: None,
            unrealised_pnl: None,
            indicative_tax: None,
            unrealised_profit: None,
            synthetic_margin: None,
            wallet_balance: None,
            margin_balance: None,
            margin_balance_pcnt: None,
            margin_leverage: None,
            margin_used_pcnt: None,
            excess_margin: None,
            excess_margin_pcnt: None,
            available_margin: None,
            withdrawable_margin: None,
            timestamp: None,
            gross_last_value: None,
            commission: None,
        }
    }

    fn build_report(
        client_order_id: &str,
        venue_order_id: &str,
        contingency_type: ContingencyType,
        order_list_id: Option<&str>,
    ) -> OrderStatusReport {
        let mut report = OrderStatusReport::new(
            AccountId::from("BITMEX-1"),
            InstrumentId::from("XBTUSD.BITMEX"),
            Some(ClientOrderId::from(client_order_id)),
            VenueOrderId::from(venue_order_id),
            OrderSide::Buy,
            OrderType::Limit,
            TimeInForce::Gtc,
            OrderStatus::Accepted,
            Quantity::new(100.0, 0),
            Quantity::default(),
            UnixNanos::from(1_u64),
            UnixNanos::from(1_u64),
            UnixNanos::from(1_u64),
            Some(UUID4::new()),
        );

        if let Some(id) = order_list_id {
            report = report.with_order_list_id(OrderListId::from(id));
        }

        report.with_contingency_type(contingency_type)
    }

    #[rstest]
    fn test_account_id_from_margins_uses_bitmex_account_number() {
        let margins = vec![margin_with_account(319111), margin_with_account(319111)];

        let account_id = account_id_from_margins(&margins).unwrap().unwrap();

        assert_eq!(account_id, AccountId::from("BITMEX-319111"));
    }

    #[rstest]
    fn test_account_id_from_margins_empty_returns_none() {
        let margins = [];

        let account_id = account_id_from_margins(&margins).unwrap();

        assert_eq!(account_id, None);
    }

    #[rstest]
    fn test_account_id_from_margins_rejects_inconsistent_accounts() {
        let margins = vec![margin_with_account(319111), margin_with_account(319112)];

        let err = account_id_from_margins(&margins).unwrap_err();

        assert!(err.to_string().contains("inconsistent margin account IDs"));
    }

    #[rstest]
    fn test_sign_request_generates_correct_headers() {
        let client = BitmexRawHttpClient::with_credentials(
            "test_api_key".to_string(),
            "test_api_secret".to_string(),
            "http://localhost:8080".to_string(),
            60,
            3,
            1_000,
            10_000,
            10_000,
            10,
            120,
            None,
        )
        .expect("Failed to create test client");

        let headers = client
            .sign_request(&Method::GET, "/api/v1/order", None)
            .unwrap();

        assert!(headers.contains_key("api-key"));
        assert!(headers.contains_key("api-signature"));
        assert!(headers.contains_key("api-expires"));
        assert_eq!(headers.get("api-key").unwrap(), "test_api_key");
    }

    #[rstest]
    fn test_sign_request_with_body() {
        let client = BitmexRawHttpClient::with_credentials(
            "test_api_key".to_string(),
            "test_api_secret".to_string(),
            "http://localhost:8080".to_string(),
            60,
            3,
            1_000,
            10_000,
            10_000,
            10,
            120,
            None,
        )
        .expect("Failed to create test client");

        let body = json!({"symbol": "XBTUSD", "orderQty": 100});
        let body_bytes = serde_json::to_vec(&body).unwrap();

        let headers_without_body = client
            .sign_request(&Method::POST, "/api/v1/order", None)
            .unwrap();
        let headers_with_body = client
            .sign_request(&Method::POST, "/api/v1/order", Some(&body_bytes))
            .unwrap();

        // Signatures should be different when body is included
        assert_ne!(
            headers_without_body.get("api-signature").unwrap(),
            headers_with_body.get("api-signature").unwrap()
        );
    }

    #[rstest]
    fn test_sign_request_uses_custom_recv_window() {
        let client_default = BitmexRawHttpClient::with_credentials(
            "test_api_key".to_string(),
            "test_api_secret".to_string(),
            "http://localhost:8080".to_string(),
            60,
            3,
            1_000,
            10_000,
            10_000, // default recv_window_ms (10000ms = 10s)
            10,
            120,
            None,
        )
        .expect("Failed to create test client");

        let client_custom = BitmexRawHttpClient::with_credentials(
            "test_api_key".to_string(),
            "test_api_secret".to_string(),
            "http://localhost:8080".to_string(),
            60,
            3,
            1_000,
            10_000,
            30_000, // 30 seconds
            10,
            120,
            None,
        )
        .expect("Failed to create test client");

        let headers_default = client_default
            .sign_request(&Method::GET, "/api/v1/order", None)
            .unwrap();
        let headers_custom = client_custom
            .sign_request(&Method::GET, "/api/v1/order", None)
            .unwrap();

        // Parse expires timestamps
        let expires_default: i64 = headers_default.get("api-expires").unwrap().parse().unwrap();
        let expires_custom: i64 = headers_custom.get("api-expires").unwrap().parse().unwrap();

        // Verify both are valid future timestamps
        let now = Utc::now().timestamp();
        assert!(expires_default > now);
        assert!(expires_custom > now);

        // Custom window should be greater than default
        assert!(expires_custom > expires_default);

        // The difference should be approximately 20 seconds (30s - 10s)
        // Allow wider tolerance for delays between calls on slow CI runners
        let diff = expires_custom - expires_default;
        assert!((18..=25).contains(&diff));
    }

    #[rstest]
    fn test_populate_linked_order_ids_from_order_list() {
        let base = "O-20250922-002219-001-000";
        let entry = format!("{base}-1");
        let stop = format!("{base}-2");
        let take = format!("{base}-3");

        let mut reports = vec![
            build_report(&entry, "V-1", ContingencyType::Oto, Some("OL-1")),
            build_report(&stop, "V-2", ContingencyType::Ouo, Some("OL-1")),
            build_report(&take, "V-3", ContingencyType::Ouo, Some("OL-1")),
        ];

        BitmexHttpClient::populate_linked_order_ids(&mut reports);

        assert_eq!(
            reports[0].linked_order_ids,
            Some(vec![
                ClientOrderId::from(stop.as_str()),
                ClientOrderId::from(take.as_str()),
            ]),
        );
        assert_eq!(
            reports[1].linked_order_ids,
            Some(vec![
                ClientOrderId::from(entry.as_str()),
                ClientOrderId::from(take.as_str()),
            ]),
        );
        assert_eq!(
            reports[2].linked_order_ids,
            Some(vec![
                ClientOrderId::from(entry.as_str()),
                ClientOrderId::from(stop.as_str()),
            ]),
        );
    }

    #[rstest]
    fn test_populate_linked_order_ids_from_id_prefix() {
        let base = "O-20250922-002220-001-000";
        let entry = format!("{base}-1");
        let stop = format!("{base}-2");
        let take = format!("{base}-3");

        let mut reports = vec![
            build_report(&entry, "V-1", ContingencyType::Oto, None),
            build_report(&stop, "V-2", ContingencyType::Ouo, None),
            build_report(&take, "V-3", ContingencyType::Ouo, None),
        ];

        BitmexHttpClient::populate_linked_order_ids(&mut reports);

        assert_eq!(
            reports[0].linked_order_ids,
            Some(vec![
                ClientOrderId::from(stop.as_str()),
                ClientOrderId::from(take.as_str()),
            ]),
        );
        assert_eq!(
            reports[1].linked_order_ids,
            Some(vec![
                ClientOrderId::from(entry.as_str()),
                ClientOrderId::from(take.as_str()),
            ]),
        );
        assert_eq!(
            reports[2].linked_order_ids,
            Some(vec![
                ClientOrderId::from(entry.as_str()),
                ClientOrderId::from(stop.as_str()),
            ]),
        );
    }

    #[rstest]
    fn test_populate_linked_order_ids_respects_non_contingent_orders() {
        let base = "O-20250922-002221-001-000";
        let entry = format!("{base}-1");
        let passive = format!("{base}-2");

        let mut reports = vec![
            build_report(&entry, "V-1", ContingencyType::NoContingency, None),
            build_report(&passive, "V-2", ContingencyType::Ouo, None),
        ];

        BitmexHttpClient::populate_linked_order_ids(&mut reports);

        // Non-contingent orders should not be linked
        assert!(reports[0].linked_order_ids.is_none());

        // A contingent order with no other contingent peers should have contingency reset
        assert!(reports[1].linked_order_ids.is_none());
        assert_eq!(reports[1].contingency_type, ContingencyType::NoContingency);
    }

    #[rstest]
    fn test_populate_linked_order_ids_treats_orphaned_oto_as_standalone() {
        let mut reports = vec![build_report(
            "O-20250922-002222-001-000-1",
            "V-1",
            ContingencyType::Oto,
            Some("OL-1"),
        )];

        BitmexHttpClient::populate_linked_order_ids(&mut reports);

        assert!(reports[0].linked_order_ids.is_none());
        assert_eq!(reports[0].contingency_type, ContingencyType::NoContingency);
        assert_eq!(reports[0].parent_order_id, None);
    }
}