nautilus_okx/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 an ergonomic wrapper around the **OKX v5 REST API** –
//! <https://www.okx.com/docs-v5/en/>.
//!
//! The core type exported by this module is [`OKXHttpClient`].  It offers an
//! interface to all exchange endpoints currently required by NautilusTrader.
//!
//! Key responsibilities handled internally:
//! • Request signing and header composition for private routes (HMAC-SHA256).
//! • Rate-limiting based on the public OKX specification.
//! • Deserialization of JSON payloads into domain models.
//! • Conversion of raw exchange errors into the rich [`OKXHttpError`] enum.
//!
//! # Official Documentation
//!
//! | Endpoint                             | Reference                                              |
//! |--------------------------------------|--------------------------------------------------------|
//! | Market data                          | <https://www.okx.com/docs-v5/en/#rest-api-market-data> |
//! | Account & positions                  | <https://www.okx.com/docs-v5/en/#rest-api-account>     |
//! | Funding & asset balances             | <https://www.okx.com/docs-v5/en/#rest-api-funding>     |

use std::{
    collections::HashMap,
    fmt::Debug,
    num::NonZeroU32,
    str::FromStr,
    sync::{
        Arc, LazyLock,
        atomic::{AtomicBool, Ordering},
    },
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use chrono::{DateTime, Utc};
use nautilus_core::{
    AtomicMap, AtomicTime, UnixNanos, consts::NAUTILUS_USER_AGENT,
    datetime::NANOSECONDS_IN_MILLISECOND, env::get_or_env_var, string::secret::REDACTED,
    time::get_atomic_clock_realtime,
};
use nautilus_model::{
    data::{
        Bar, BarType, BookOrder, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate,
        OrderBookDelta, OrderBookDeltas, TradeTick, forward::ForwardPrice,
    },
    enums::{
        AggregationSource, BarAggregation, BookAction, BookType, OrderSide, OrderType,
        PositionSide, RecordFlag, TimeInForce, TriggerType,
    },
    events::AccountState,
    identifiers::{AccountId, ClientOrderId, InstrumentId},
    instruments::{Instrument, InstrumentAny},
    orderbook::OrderBook,
    reports::{FillReport, OrderStatusReport, PositionStatusReport},
    types::{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 tokio_util::sync::CancellationToken;
use ustr::Ustr;

use super::{
    error::OKXHttpError,
    models::{
        OKXAccount, OKXAmendAlgoOrderRequest, OKXAmendAlgoOrderResponse, OKXAttachAlgoOrdRequest,
        OKXCancelAlgoOrderRequest, OKXCancelAlgoOrderResponse, OKXFeeRate, OKXFundingRateHistory,
        OKXIndexTicker, OKXMarkPrice, OKXOptionSummary, OKXOrderAlgo, OKXOrderBookSnapshot,
        OKXOrderHistory, OKXPlaceAlgoOrderRequest, OKXPlaceAlgoOrderResponse, OKXPlaceOrderRequest,
        OKXPlaceOrderResponse, OKXPosition, OKXPositionHistory, OKXPositionTier, OKXServerTime,
        OKXTransactionDetail,
    },
    query::{
        GetAlgoOrdersParams, GetAlgoOrdersParamsBuilder, GetCandlesticksParams,
        GetCandlesticksParamsBuilder, GetFundingRateHistoryParams, GetIndexTickerParams,
        GetIndexTickerParamsBuilder, GetInstrumentsParams, GetInstrumentsParamsBuilder,
        GetMarkPriceParams, GetMarkPriceParamsBuilder, GetOptionSummaryParams, GetOrderBookParams,
        GetOrderHistoryParams, GetOrderHistoryParamsBuilder, GetOrderListParams,
        GetOrderListParamsBuilder, GetPositionTiersParams, GetPositionsHistoryParams,
        GetPositionsParams, GetPositionsParamsBuilder, GetTradeFeeParams, GetTradesParams,
        GetTradesParamsBuilder, GetTransactionDetailsParams, GetTransactionDetailsParamsBuilder,
        SetPositionModeParams, SetPositionModeParamsBuilder,
    },
};
use crate::{
    common::{
        consts::{
            OKX_FIELD_SCODE, OKX_FIELD_SMSG, OKX_HTTP_URL, OKX_NAUTILUS_BROKER_ID,
            OKX_SUPPORTED_ORDER_TYPES, OKX_SUPPORTED_TIME_IN_FORCE, should_retry_error_code,
        },
        credential::Credential,
        enums::{
            OKXAlgoOrderType, OKXContractType, OKXEnvironment, OKXInstrumentStatus,
            OKXInstrumentType, OKXOrderStatus, OKXOrderType, OKXPositionMode, OKXPositionSide,
            OKXSide, OKXTargetCurrency, OKXTradeMode, OKXTriggerType,
            conditional_order_to_algo_type,
        },
        models::OKXInstrument,
        parse::{
            extract_inst_family, okx_instrument_type, okx_instrument_type_from_symbol,
            parse_account_state, parse_base_quote_from_symbol, parse_candlestick,
            parse_fill_report, parse_funding_rate, parse_index_price_update, parse_instrument_any,
            parse_instrument_id, parse_mark_price_update, parse_order_status_report,
            parse_position_status_report, parse_price, parse_quantity,
            parse_spot_margin_position_from_balance, parse_trade_tick,
        },
    },
    http::{
        models::{OKXCandlestick, OKXTrade},
        query::GetOrderParams,
    },
    websocket::{messages::OKXAlgoOrderMsg, parse::parse_algo_order_status_report},
};

const OKX_SUCCESS_CODE: &str = "0";

/// Ranks a spot instrument's quote currency for deterministic tie-breaking
/// when multiple pairs share the same base. Matches OKX's dominant-quote
/// ordering so spot-margin position reports stay on a stable instrument id
/// across restarts.
fn spot_quote_priority(symbol: &str) -> u8 {
    symbol.rsplit_once('-').map_or(4, |(_, quote)| match quote {
        "USDT" => 0,
        "USDC" => 1,
        "USD" => 2,
        _ => 3,
    })
}

fn resolve_okx_error_message(response_body: &[u8], top_level_msg: &str) -> String {
    let message = top_level_msg.trim();
    let is_generic_top_level = message.eq_ignore_ascii_case("All operations failed");
    if !message.is_empty() && !is_generic_top_level {
        return message.to_string();
    }

    if let Ok(payload) = serde_json::from_slice::<serde_json::Value>(response_body)
        && let Some(first_item) = payload
            .get("data")
            .and_then(serde_json::Value::as_array)
            .and_then(|items| items.first())
    {
        if let Some(s_msg) = first_item
            .get(OKX_FIELD_SMSG)
            .and_then(serde_json::Value::as_str)
        {
            let s_msg = s_msg.trim();
            if !s_msg.is_empty() {
                return s_msg.to_string();
            }
        }

        if let Some(s_code) = first_item
            .get(OKX_FIELD_SCODE)
            .and_then(serde_json::Value::as_str)
        {
            let s_code = s_code.trim();
            if !s_code.is_empty() {
                return s_code.to_string();
            }
        }
    }

    String::new()
}

#[cfg(test)]
mod tests {
    use rstest::rstest;

    use super::resolve_okx_error_message;

    #[rstest]
    fn test_resolve_okx_error_message_prefers_detailed_s_msg_over_generic_top_level() {
        let body = br#"{
            "code": "1",
            "msg": "All operations failed",
            "data": [
                {
                    "sCode": "51046",
                    "sMsg": "Test detailed failure"
                }
            ]
        }"#;

        assert_eq!(
            resolve_okx_error_message(body, "All operations failed"),
            "Test detailed failure",
        );
    }

    #[rstest]
    #[case("BTC-USD")]
    #[case("BTC-USD-241217")]
    #[case("BTC-USD-241217-92000")]
    fn test_option_summary_expiry_key_rejects_short_symbol(#[case] symbol: &str) {
        let result = super::OKXHttpClient::option_summary_expiry_key(symbol);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("Expected OKX option symbol with expiry"),
            "unexpected error: {err}"
        );
    }

    #[rstest]
    fn test_option_summary_expiry_key_extracts_base_quote_expiry() {
        let result =
            super::OKXHttpClient::option_summary_expiry_key("BTC-USD-241217-92000-C").unwrap();
        assert_eq!(result, "BTC-USD-241217");
    }

    #[rstest]
    #[case("BTC-USD")]
    #[case("BTC-USD-241217")]
    #[case("BTC-USD-241217-92000")]
    fn test_option_summary_exp_time_rejects_short_symbol(#[case] symbol: &str) {
        let result = super::OKXHttpClient::option_summary_exp_time(symbol);
        assert!(result.is_err());
        let err = result.unwrap_err().to_string();
        assert!(
            err.contains("Expected OKX option symbol with expiry"),
            "unexpected error: {err}"
        );
    }

    #[rstest]
    fn test_option_summary_exp_time_extracts_expiry() {
        let result =
            super::OKXHttpClient::option_summary_exp_time("BTC-USD-241217-92000-C").unwrap();
        assert_eq!(result, Some("241217".to_string()));
    }
}

/// Default OKX REST API rate limit: 500 requests per 2 seconds.
///
/// - Sub-account order limit: 1000 requests per 2 seconds.
/// - Account balance: 10 requests per 2 seconds.
/// - Account instruments: 20 requests per 2 seconds.
///
/// We use a conservative 250 requests per second (500 per 2 seconds) as a general limit
/// that should accommodate most use cases while respecting OKX's documented limits.
pub static OKX_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
    Quota::per_second(NonZeroU32::new(250).expect("non-zero")).expect("valid constant")
});

const OKX_GLOBAL_RATE_KEY: &str = "okx:global";

// OKX returns at most 100 records per page for order, fill, and algo endpoints
const OKX_PAGE_SIZE: usize = 100;

// Safety cap on paginated reconciliation fetches to avoid unbounded loops
const MAX_RECONCILIATION_PAGES: usize = 50;

/// Represents an OKX HTTP response.
#[derive(Debug, Serialize, Deserialize)]
pub struct OKXResponse<T> {
    /// The OKX response code, which is `"0"` for success.
    pub code: String,
    /// A message string which can be informational or describe an error cause.
    pub msg: String,
    /// The typed data returned by the OKX endpoint.
    pub data: Vec<T>,
}

/// Provides a raw HTTP client for interacting with the [OKX](https://okx.com) REST API.
///
/// This client wraps the underlying [`HttpClient`] to handle functionality
/// specific to OKX, such as request signing (for authenticated endpoints),
/// forming request URLs, and deserializing responses into OKX specific data models.
pub struct OKXRawHttpClient {
    base_url: String,
    client: HttpClient,
    credential: Option<Credential>,
    retry_manager: RetryManager<OKXHttpError>,
    cancellation_token: CancellationToken,
    environment: OKXEnvironment,
}

impl Default for OKXRawHttpClient {
    fn default() -> Self {
        Self::new(None, 60, 3, 1000, 10_000, OKXEnvironment::Live, None)
            .expect("Failed to create default OKXRawHttpClient")
    }
}

impl Debug for OKXRawHttpClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let credential = self.credential.as_ref().map(|_| REDACTED);
        f.debug_struct(stringify!(OKXRawHttpClient))
            .field("base_url", &self.base_url)
            .field("credential", &credential)
            .finish_non_exhaustive()
    }
}

impl OKXRawHttpClient {
    fn rate_limiter_quotas() -> Vec<(String, Quota)> {
        vec![
            (OKX_GLOBAL_RATE_KEY.to_string(), *OKX_REST_QUOTA),
            (
                "okx:/api/v5/account/balance".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/instruments".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/candles".to_string(),
                Quota::per_second(NonZeroU32::new(50).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/history-candles".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/history-trades".to_string(),
                Quota::per_second(NonZeroU32::new(30).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/order".to_string(),
                Quota::per_second(NonZeroU32::new(30).expect("non-zero")).expect("valid constant"), // 60 requests / 2 seconds (per instrument)
            ),
            (
                "okx:/api/v5/trade/orders-pending".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/orders-history".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/fills".to_string(),
                Quota::per_second(NonZeroU32::new(30).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/order-algo".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/cancel-algos".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/amend-algos".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
        ]
    }

    fn rate_limit_keys(endpoint: &str) -> Vec<Ustr> {
        let normalized = endpoint.split('?').next().unwrap_or(endpoint);
        let route = format!("okx:{normalized}");

        vec![Ustr::from(OKX_GLOBAL_RATE_KEY), Ustr::from(route.as_str())]
    }

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

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

    /// Creates a new [`OKXHttpClient`] using the default OKX 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.
    pub fn new(
        base_url: Option<String>,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        environment: OKXEnvironment,
        proxy_url: Option<String>,
    ) -> Result<Self, OKXHttpError> {
        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(OKX_HTTP_URL.to_string()),
            client: HttpClient::new(
                Self::default_headers(environment),
                vec![],
                Self::rate_limiter_quotas(),
                Some(*OKX_REST_QUOTA),
                Some(timeout_secs),
                proxy_url,
            )
            .map_err(|e| {
                OKXHttpError::ValidationError(format!("Failed to create HTTP client: {e}"))
            })?,
            credential: None,
            retry_manager,
            cancellation_token: CancellationToken::new(),
            environment,
        })
    }

    /// Creates a new [`OKXHttpClient`] 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,
        api_passphrase: String,
        base_url: String,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        environment: OKXEnvironment,
        proxy_url: Option<String>,
    ) -> Result<Self, OKXHttpError> {
        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(environment),
                vec![],
                Self::rate_limiter_quotas(),
                Some(*OKX_REST_QUOTA),
                Some(timeout_secs),
                proxy_url,
            )
            .map_err(|e| {
                OKXHttpError::ValidationError(format!("Failed to create HTTP client: {e}"))
            })?,
            credential: Some(Credential::new(api_key, api_secret, api_passphrase)),
            retry_manager,
            cancellation_token: CancellationToken::new(),
            environment,
        })
    }

    /// Builds the default headers to include with each request (e.g., `User-Agent`).
    fn default_headers(environment: OKXEnvironment) -> HashMap<String, String> {
        let mut headers =
            HashMap::from([(USER_AGENT.to_string(), NAUTILUS_USER_AGENT.to_string())]);

        if environment == OKXEnvironment::Demo {
            headers.insert("x-simulated-trading".to_string(), "1".to_string());
        }

        headers
    }

    /// Signs an OKX request with timestamp, API key, passphrase, and signature.
    ///
    /// # Errors
    ///
    /// Returns [`OKXHttpError::MissingCredentials`] if no credentials are set
    /// but the request requires authentication.
    fn sign_request(
        &self,
        method: &Method,
        path: &str,
        body: Option<&[u8]>,
    ) -> Result<HashMap<String, String>, OKXHttpError> {
        let credential = match self.credential.as_ref() {
            Some(c) => c,
            None => return Err(OKXHttpError::MissingCredentials),
        };

        let api_key = credential.api_key().to_string();
        let api_passphrase = credential.api_passphrase().to_string();

        // OKX requires milliseconds in the timestamp (ISO 8601 with milliseconds)
        let now = Utc::now();
        let millis = now.timestamp_subsec_millis();
        let timestamp = now.format("%Y-%m-%dT%H:%M:%S").to_string() + &format!(".{millis:03}Z");
        let signature = credential.sign_bytes(&timestamp, method.as_str(), path, body);

        let mut headers = HashMap::new();
        headers.insert("OK-ACCESS-KEY".to_string(), api_key);
        headers.insert("OK-ACCESS-PASSPHRASE".to_string(), api_passphrase);
        headers.insert("OK-ACCESS-TIMESTAMP".to_string(), timestamp);
        headers.insert("OK-ACCESS-SIGN".to_string(), signature);

        Ok(headers)
    }

    /// Sends an HTTP request to OKX and parses the response into `Vec<T>`.
    ///
    /// Internally, this method handles:
    /// - Building the URL from `base_url` + `path`.
    /// - Optionally signing the request.
    /// - Deserializing JSON responses into typed models, or returning a [`OKXHttpError`].
    /// - Retrying with exponential backoff on transient errors.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The HTTP request fails.
    /// - Authentication is required but credentials are missing.
    /// - The response cannot be deserialized into the expected type.
    /// - The OKX API returns an error response.
    async fn send_request<T: DeserializeOwned, P: Serialize>(
        &self,
        method: Method,
        path: &str,
        params: Option<&P>,
        body: Option<Vec<u8>>,
        authenticate: bool,
    ) -> Result<Vec<T>, OKXHttpError> {
        let url = format!("{}{path}", self.base_url);

        // Pre-compute rate limit keys once outside the retry closure
        let rate_keys: Vec<String> = Self::rate_limit_keys(path)
            .into_iter()
            .map(|k| k.to_string())
            .collect();

        let operation = || {
            let url = url.clone();
            let method = method.clone();
            let body = body.clone();
            let rate_keys = rate_keys.clone();

            async move {
                // Serialize params to query string for signing (if needed)
                let query_string = if let Some(p) = params {
                    serde_urlencoded::to_string(p).map_err(|e| {
                        OKXHttpError::JsonError(format!("Failed to serialize params: {e}"))
                    })?
                } else {
                    String::new()
                };

                // Build full path with query string for signing
                let full_path = if query_string.is_empty() {
                    path.to_string()
                } else {
                    format!("{path}?{query_string}")
                };

                let mut headers = if authenticate {
                    self.sign_request(&method, &full_path, body.as_deref())?
                } else {
                    HashMap::new()
                };

                // Always set Content-Type header when body is present
                if body.is_some() {
                    headers.insert("Content-Type".to_string(), "application/json".to_string());
                }

                let resp = self
                    .client
                    .request_with_params(
                        method.clone(),
                        url,
                        params,
                        Some(headers),
                        body,
                        None,
                        Some(rate_keys),
                    )
                    .await?;

                log::trace!("Response: {resp:?}");

                if resp.status.is_success() {
                    let okx_response: OKXResponse<T> =
                        serde_json::from_slice(&resp.body).map_err(|e| {
                            log::error!("Failed to deserialize OKXResponse: {e}");
                            OKXHttpError::JsonError(e.to_string())
                        })?;

                    if okx_response.code != OKX_SUCCESS_CODE {
                        return Err(OKXHttpError::OkxError {
                            error_code: okx_response.code,
                            message: resolve_okx_error_message(&resp.body, &okx_response.msg),
                        });
                    }

                    Ok(okx_response.data)
                } else {
                    let error_body = String::from_utf8_lossy(&resp.body);
                    if resp.status.as_u16() == StatusCode::NOT_FOUND.as_u16() {
                        log::debug!("HTTP 404 with body: {error_body}");
                    } else {
                        log::error!(
                            "HTTP error {} with body: {error_body}",
                            resp.status.as_str()
                        );
                    }

                    if let Ok(parsed_error) = serde_json::from_slice::<OKXResponse<T>>(&resp.body) {
                        return Err(OKXHttpError::OkxError {
                            error_code: parsed_error.code,
                            message: resolve_okx_error_message(&resp.body, &parsed_error.msg),
                        });
                    }

                    Err(OKXHttpError::UnexpectedStatus {
                        // Fall back to 500 if the venue returns a non-standard
                        // code so we never panic in the error path.
                        status: StatusCode::from_u16(resp.status.as_u16())
                            .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
                        body: error_body.to_string(),
                    })
                }
            }
        };

        // Retry strategy based on OKX 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. OKX specific retryable error codes (defined in common::consts)
        //
        // Note: OKX returns many permanent errors which should NOT be retried
        // (e.g., "Invalid instrument", "Insufficient balance", "Invalid API Key")
        let should_retry = |error: &OKXHttpError| -> bool {
            match error {
                OKXHttpError::HttpClientError(_) => true,
                OKXHttpError::UnexpectedStatus { status, .. } => {
                    status.as_u16() >= 500 || status.as_u16() == 429
                }
                OKXHttpError::OkxError { error_code, .. } => should_retry_error_code(error_code),
                _ => false,
            }
        };

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

        self.retry_manager
            .execute_with_retry_with_cancel(
                path,
                operation,
                should_retry,
                create_error,
                &self.cancellation_token,
            )
            .await
    }

    /// Sets the position mode for an account.
    ///
    /// # Errors
    ///
    /// Returns an error if JSON serialization of `params` fails, if the HTTP
    /// request fails, or if the response body cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-set-position-mode>
    pub async fn set_position_mode(
        &self,
        params: SetPositionModeParams,
    ) -> Result<Vec<serde_json::Value>, OKXHttpError> {
        let path = "/api/v5/account/set-position-mode";
        let body = serde_json::to_vec(&params)?;
        self.send_request::<_, ()>(Method::POST, path, None, Some(body), true)
            .await
    }

    /// Requests position tiers information, maximum leverage depends on your borrowings and margin ratio.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails, authentication is rejected
    /// or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-position-tiers>
    pub async fn get_position_tiers(
        &self,
        params: GetPositionTiersParams,
    ) -> Result<Vec<OKXPositionTier>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/position-tiers",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests a list of instruments with open contracts.
    ///
    /// # Errors
    ///
    /// Returns an error if JSON serialization of `params` fails, if the HTTP
    /// request fails, or if the response body cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-instruments>
    pub async fn get_instruments(
        &self,
        params: GetInstrumentsParams,
    ) -> Result<Vec<OKXInstrument>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/instruments",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests option market data for an instrument family.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-option-market-data>
    pub async fn get_option_summary(
        &self,
        params: GetOptionSummaryParams,
    ) -> Result<Vec<OKXOptionSummary>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/opt-summary",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests the current server time from OKX.
    ///
    /// Retrieves the OKX 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 [`OKXServerTime`].
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-system-time>
    pub async fn get_server_time(&self) -> Result<u64, OKXHttpError> {
        let response: Vec<OKXServerTime> = self
            .send_request::<_, ()>(Method::GET, "/api/v5/public/time", None, None, false)
            .await?;
        response
            .first()
            .map(|t| t.ts)
            .ok_or_else(|| OKXHttpError::JsonError("Empty server time response".to_string()))
    }

    /// Requests a mark price.
    ///
    /// We set the mark price based on the SPOT index and at a reasonable basis to prevent individual
    /// users from manipulating the market and causing the contract price to fluctuate.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or if the response body
    /// cannot be parsed into [`OKXMarkPrice`].
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-mark-price>
    pub async fn get_mark_price(
        &self,
        params: GetMarkPriceParams,
    ) -> Result<Vec<OKXMarkPrice>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/mark-price",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests the latest index price.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-index-tickers>
    pub async fn get_index_tickers(
        &self,
        params: GetIndexTickerParams,
    ) -> Result<Vec<OKXIndexTicker>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/index-tickers",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests trades history.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-market-data-get-trades-history>
    pub async fn get_history_trades(
        &self,
        params: GetTradesParams,
    ) -> Result<Vec<OKXTrade>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/history-trades",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests order book snapshot.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-market-data-get-order-book>
    pub async fn get_order_book(
        &self,
        params: GetOrderBookParams,
    ) -> Result<Vec<OKXOrderBookSnapshot>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/books",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests funding rate history.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-funding-rate-history>
    pub async fn get_funding_rate_history(
        &self,
        params: GetFundingRateHistoryParams,
    ) -> Result<Vec<OKXFundingRateHistory>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/funding-rate-history",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests recent candlestick data.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-market-data-get-candlesticks>
    pub async fn get_candles(
        &self,
        params: GetCandlesticksParams,
    ) -> Result<Vec<OKXCandlestick>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/candles",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests historical candlestick data.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-market-data-get-candlesticks-history>
    pub async fn get_history_candles(
        &self,
        params: GetCandlesticksParams,
    ) -> Result<Vec<OKXCandlestick>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/history-candles",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests a list of assets (with non-zero balance), remaining balance, and available amount
    /// in the trading account.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-balance>
    pub async fn get_balance(&self) -> Result<Vec<OKXAccount>, OKXHttpError> {
        let path = "/api/v5/account/balance";
        self.send_request::<_, ()>(Method::GET, path, None, None, true)
            .await
    }

    /// Requests fee rates for the account.
    ///
    /// Returns fee rates for the specified instrument type and the user's VIP level.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-fee-rates>
    pub async fn get_trade_fee(
        &self,
        params: GetTradeFeeParams,
    ) -> Result<Vec<OKXFeeRate>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/trade-fee",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Retrieves a single order’s details.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-order>
    pub async fn get_order(
        &self,
        params: GetOrderParams,
    ) -> Result<Vec<OKXOrderHistory>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/order",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests order list (pending orders).
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-order-list>
    pub async fn get_orders_pending(
        &self,
        params: GetOrderListParams,
    ) -> Result<Vec<OKXOrderHistory>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/orders-pending",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests historical order records.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-orders-history>
    pub async fn get_orders_history(
        &self,
        params: GetOrderHistoryParams,
    ) -> Result<Vec<OKXOrderHistory>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/orders-history",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests pending algo orders.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub async fn get_order_algo_pending(
        &self,
        params: GetAlgoOrdersParams,
    ) -> Result<Vec<OKXOrderAlgo>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/orders-algo-pending",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests historical algo orders.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub async fn get_order_algo_history(
        &self,
        params: GetAlgoOrdersParams,
    ) -> Result<Vec<OKXOrderAlgo>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/orders-algo-history",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests transaction details (fills) for the given parameters.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-transaction-details-last-3-days>
    pub async fn get_fills(
        &self,
        params: GetTransactionDetailsParams,
    ) -> Result<Vec<OKXTransactionDetail>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/fills",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests information on your positions. When the account is in net mode, net positions will
    /// be displayed, and when the account is in long/short mode, long or short positions will be
    /// displayed. Returns in reverse chronological order using ctime.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-positions>
    pub async fn get_positions(
        &self,
        params: GetPositionsParams,
    ) -> Result<Vec<OKXPosition>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/positions",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests closed or historical position data.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-positions-history>
    pub async fn get_positions_history(
        &self,
        params: GetPositionsHistoryParams,
    ) -> Result<Vec<OKXPositionHistory>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/positions-history",
            Some(&params),
            None,
            true,
        )
        .await
    }
}

/// Provides a higher-level HTTP client for the [OKX](https://okx.com) REST API.
///
/// This client wraps the underlying `OKXHttpInnerClient` to handle conversions
/// into the Nautilus domain model.
#[derive(Debug)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.okx", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.okx")
)]
pub struct OKXHttpClient {
    pub(crate) inner: Arc<OKXRawHttpClient>,
    pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
    clock: &'static AtomicTime,
    cache_initialized: AtomicBool,
}

impl Clone for OKXHttpClient {
    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(),
            cache_initialized,
            clock: self.clock,
        }
    }
}

impl Default for OKXHttpClient {
    fn default() -> Self {
        Self::new(None, 60, 3, 1000, 10_000, OKXEnvironment::Live, None)
            .expect("Failed to create default OKXHttpClient")
    }
}

impl OKXHttpClient {
    /// Creates a new [`OKXHttpClient`] using the default OKX 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.
    pub fn new(
        base_url: Option<String>,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        environment: OKXEnvironment,
        proxy_url: Option<String>,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            inner: Arc::new(OKXRawHttpClient::new(
                base_url,
                timeout_secs,
                max_retries,
                retry_delay_ms,
                retry_delay_max_ms,
                environment,
                proxy_url,
            )?),
            instruments_cache: Arc::new(AtomicMap::new()),
            cache_initialized: AtomicBool::new(false),
            clock: get_atomic_clock_realtime(),
        })
    }

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

    /// Creates a new authenticated [`OKXHttpClient`] using environment variables and
    /// the default OKX HTTP base url.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub fn from_env() -> anyhow::Result<Self> {
        Self::with_credentials(
            None,
            None,
            None,
            None,
            60,
            3,
            1000,
            10_000,
            OKXEnvironment::Live,
            None,
        )
    }

    /// Creates a new [`OKXHttpClient`] configured with credentials
    /// for authenticated requests, optionally using a custom base url.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    #[expect(clippy::too_many_arguments)]
    pub fn with_credentials(
        api_key: Option<String>,
        api_secret: Option<String>,
        api_passphrase: Option<String>,
        base_url: Option<String>,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
        retry_delay_max_ms: u64,
        environment: OKXEnvironment,
        proxy_url: Option<String>,
    ) -> anyhow::Result<Self> {
        let api_key = get_or_env_var(api_key, "OKX_API_KEY")?;
        let api_secret = get_or_env_var(api_secret, "OKX_API_SECRET")?;
        let api_passphrase = get_or_env_var(api_passphrase, "OKX_API_PASSPHRASE")?;
        let base_url = base_url.unwrap_or(OKX_HTTP_URL.to_string());

        Ok(Self {
            inner: Arc::new(OKXRawHttpClient::with_credentials(
                api_key,
                api_secret,
                api_passphrase,
                base_url,
                timeout_secs,
                max_retries,
                retry_delay_ms,
                retry_delay_max_ms,
                environment,
                proxy_url,
            )?),
            instruments_cache: Arc::new(AtomicMap::new()),
            cache_initialized: AtomicBool::new(false),
            clock: get_atomic_clock_realtime(),
        })
    }

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

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

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

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

    /// Returns the public API key being used by the client.
    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())
    }

    /// Returns whether the client is configured for demo trading.
    #[must_use]
    pub fn is_demo(&self) -> bool {
        self.inner.environment == OKXEnvironment::Demo
    }

    /// Requests the current server time from OKX.
    ///
    /// Returns the OKX 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, OKXHttpError> {
        self.inner.get_server_time().await
    }

    /// Checks if the client is initialized.
    ///
    /// The client is considered initialized if any instruments have been cached from the venue.
    #[must_use]
    pub fn is_initialized(&self) -> bool {
        self.cache_initialized.load(Ordering::Acquire)
    }

    /// Returns a snapshot of all instrument symbols currently held in the
    /// internal cache.
    #[must_use]
    pub fn get_cached_symbols(&self) -> Vec<String> {
        self.instruments_cache
            .load()
            .keys()
            .map(|k| k.to_string())
            .collect()
    }

    /// 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);
    }

    /// 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);
    }

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

    /// Requests the account state for the `account_id` from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or no account state is returned.
    pub async fn request_account_state(
        &self,
        account_id: AccountId,
    ) -> anyhow::Result<AccountState> {
        let resp = self
            .inner
            .get_balance()
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let ts_init = self.generate_ts_init();
        let raw = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No account state returned from OKX"))?;
        let account_state = parse_account_state(raw, account_id, ts_init)?;

        Ok(account_state)
    }

    /// Sets the position mode for the account.
    ///
    /// Defaults to NetMode if no position mode is provided.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the position mode cannot be set.
    ///
    /// # Note
    ///
    /// This endpoint only works for accounts with derivatives trading enabled.
    /// If the account only has spot trading, this will return an error.
    pub async fn set_position_mode(&self, position_mode: OKXPositionMode) -> anyhow::Result<()> {
        let mut params = SetPositionModeParamsBuilder::default();
        params.pos_mode(position_mode);
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        match self.inner.set_position_mode(params).await {
            Ok(_) => Ok(()),
            Err(e) => {
                if let OKXHttpError::OkxError {
                    error_code,
                    message,
                } = &e
                    && error_code == "50115"
                {
                    log::warn!(
                        "Account does not support position mode setting (derivatives trading not enabled): {message}"
                    );
                    return Ok(()); // Gracefully handle this case
                }
                anyhow::bail!(e)
            }
        }
    }

    /// Requests all instruments for the `instrument_type` from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or instrument parsing fails.
    ///
    /// # Returns
    ///
    /// A tuple containing:
    /// - `Vec<InstrumentAny>`: The parsed instruments
    /// - `Vec<(Ustr, u64)>`: Mappings of inst_id to inst_id_code for WebSocket order operations
    pub async fn request_instruments(
        &self,
        instrument_type: OKXInstrumentType,
        instrument_family: Option<String>,
    ) -> anyhow::Result<(Vec<InstrumentAny>, Vec<(Ustr, u64)>)> {
        let mut params = GetInstrumentsParamsBuilder::default();
        params.inst_type(instrument_type);

        if let Some(family) = instrument_family.clone() {
            params.inst_family(family);
        }

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

        let resp = self
            .inner
            .get_instruments(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let fee_rate_opt = {
            let fee_params = GetTradeFeeParams {
                inst_type: instrument_type,
                uly: None,
                inst_family: instrument_family,
            };

            match self.inner.get_trade_fee(fee_params).await {
                Ok(rates) => rates.into_iter().next(),
                Err(OKXHttpError::MissingCredentials) => {
                    log::debug!("Missing credentials for fee rates, using None");
                    None
                }
                Err(e) => {
                    log::warn!("Failed to fetch fee rates for {instrument_type}: {e}");
                    None
                }
            }
        };

        let ts_init = self.generate_ts_init();

        let mut instruments: Vec<InstrumentAny> = Vec::new();
        let mut inst_id_codes: Vec<(Ustr, u64)> = Vec::new();

        for inst in &resp {
            // Collect inst_id_code mappings for WebSocket order operations
            if let Some(code) = inst.inst_id_code {
                inst_id_codes.push((inst.inst_id, code));
            }
            // Skip pre-open instruments which have incomplete/empty field values
            // Keep suspended instruments as they have valid metadata and may return to live
            if inst.state == OKXInstrumentStatus::Preopen {
                continue;
            }

            // Determine which fee fields to use based on contract type
            // OKX fee rate convention: positive = rebate, negative = commission
            // Nautilus convention: negative = rebate, positive = commission
            // Negate to convert between conventions
            let (maker_fee, taker_fee) = if let Some(ref fee_rate) = fee_rate_opt {
                let is_usdt_margined = inst.ct_type == OKXContractType::Linear;
                let (maker_str, taker_str) = if is_usdt_margined {
                    (&fee_rate.maker_u, &fee_rate.taker_u)
                } else {
                    (&fee_rate.maker, &fee_rate.taker)
                };

                let maker = if maker_str.is_empty() {
                    None
                } else {
                    Decimal::from_str(maker_str).ok().map(|v| -v)
                };
                let taker = if taker_str.is_empty() {
                    None
                } else {
                    Decimal::from_str(taker_str).ok().map(|v| -v)
                };

                (maker, taker)
            } else {
                (None, None)
            };

            match parse_instrument_any(inst, None, None, maker_fee, taker_fee, ts_init) {
                Ok(Some(instrument_any)) => {
                    instruments.push(instrument_any);
                }
                Ok(None) => {
                    // Unsupported instrument type, skip silently
                }
                Err(e) => {
                    log::warn!("Failed to parse instrument {}: {e}", inst.inst_id);
                }
            }
        }

        Ok((instruments, inst_id_codes))
    }

    /// Requests a single instrument by `instrument_id` from OKX.
    ///
    /// Fetches the instrument from the API, caches it, and returns it.
    ///
    /// # Errors
    ///
    /// This function will return an error if:
    /// - The API request fails.
    /// - The instrument is not found.
    /// - Failed to parse instrument data.
    pub async fn request_instrument(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<InstrumentAny> {
        let symbol = instrument_id.symbol.as_str();
        let instrument_type = okx_instrument_type_from_symbol(symbol);

        let mut params = GetInstrumentsParamsBuilder::default();
        params.inst_type(instrument_type);
        params.inst_id(symbol);

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

        let resp = self
            .inner
            .get_instruments(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let raw_inst = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("Instrument {symbol} not found"))?;

        // Skip pre-open instruments which have incomplete/empty field values
        if raw_inst.state == OKXInstrumentStatus::Preopen {
            anyhow::bail!("Instrument {symbol} is in pre-open state");
        }

        let fee_rate_opt = {
            let fee_params = GetTradeFeeParams {
                inst_type: instrument_type,
                uly: None,
                inst_family: None,
            };

            match self.inner.get_trade_fee(fee_params).await {
                Ok(rates) => rates.into_iter().next(),
                Err(OKXHttpError::MissingCredentials) => {
                    log::debug!("Missing credentials for fee rates, using None");
                    None
                }
                Err(e) => {
                    log::warn!("Failed to fetch fee rates for {symbol}: {e}");
                    None
                }
            }
        };

        // OKX fee rate convention: positive = rebate, negative = commission
        // Nautilus convention: negative = rebate, positive = commission
        // Negate to convert between conventions
        let (maker_fee, taker_fee) = if let Some(ref fee_rate) = fee_rate_opt {
            let is_usdt_margined = raw_inst.ct_type == OKXContractType::Linear;
            let (maker_str, taker_str) = if is_usdt_margined {
                (&fee_rate.maker_u, &fee_rate.taker_u)
            } else {
                (&fee_rate.maker, &fee_rate.taker)
            };

            let maker = if maker_str.is_empty() {
                None
            } else {
                Decimal::from_str(maker_str).ok().map(|v| -v)
            };
            let taker = if taker_str.is_empty() {
                None
            } else {
                Decimal::from_str(taker_str).ok().map(|v| -v)
            };

            (maker, taker)
        } else {
            (None, None)
        };

        let ts_init = self.generate_ts_init();
        let instrument = parse_instrument_any(raw_inst, None, None, maker_fee, taker_fee, ts_init)?
            .ok_or_else(|| anyhow::anyhow!("Unsupported instrument type for {symbol}"))?;

        self.cache_instrument(instrument.clone());

        Ok(instrument)
    }

    /// Requests forward prices for OKX options using the option summary endpoint.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or no usable instrument family can be resolved.
    pub async fn request_forward_prices(
        &self,
        underlying: &str,
        instrument_id: Option<InstrumentId>,
    ) -> anyhow::Result<Vec<ForwardPrice>> {
        let requests = self.resolve_forward_price_requests(underlying, instrument_id.as_ref())?;
        let requested_symbol = instrument_id.as_ref().map(|id| id.symbol.inner());
        let requested_instrument_id = instrument_id.as_ref();
        let ts_init = self.generate_ts_init();
        let mut forward_prices = Vec::new();
        let mut seen_expiries = AHashSet::new();

        for (inst_family, exp_time) in requests {
            let summaries = self
                .inner
                .get_option_summary(GetOptionSummaryParams {
                    inst_family,
                    exp_time,
                })
                .await
                .map_err(|e| anyhow::anyhow!(e))?;

            for summary in summaries {
                if summary.inst_type != OKXInstrumentType::Option {
                    continue;
                }

                if let Some(symbol) = requested_symbol
                    && summary.inst_id != symbol
                {
                    continue;
                }

                let forward_price = match Decimal::from_str(&summary.fwd_px) {
                    Ok(price) if !price.is_zero() => price,
                    Ok(_) => continue,
                    Err(e) => {
                        log::warn!(
                            "Skipping invalid OKX forward price for {}: {e}",
                            summary.inst_id
                        );
                        continue;
                    }
                };

                if requested_symbol.is_none() {
                    let expiry_key = Self::option_summary_expiry_key(summary.inst_id.as_str())?;
                    if !seen_expiries.insert(expiry_key) {
                        continue;
                    }
                }

                let ts_event =
                    UnixNanos::from(summary.ts.saturating_mul(NANOSECONDS_IN_MILLISECOND));
                let instrument_id = if let Some(inst_id) = requested_instrument_id {
                    *inst_id
                } else {
                    parse_instrument_id(summary.inst_id)
                };

                forward_prices.push(ForwardPrice::new(
                    instrument_id,
                    forward_price,
                    Some(summary.uly.to_string()),
                    ts_event,
                    ts_init,
                ));
            }
        }

        Ok(forward_prices)
    }

    /// Requests the latest mark price for the `instrument_type` from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or no mark price is returned.
    pub async fn request_mark_price(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<MarkPriceUpdate> {
        let mut params = GetMarkPriceParamsBuilder::default();
        params.inst_id(instrument_id.symbol.inner());
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let resp = self
            .inner
            .get_mark_price(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let raw = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No mark price returned from OKX"))?;
        let inst = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        let mark_price =
            parse_mark_price_update(raw, instrument_id, inst.price_precision(), ts_init)
                .map_err(|e| anyhow::anyhow!(e))?;
        Ok(mark_price)
    }

    fn resolve_forward_price_requests(
        &self,
        underlying: &str,
        instrument_id: Option<&InstrumentId>,
    ) -> anyhow::Result<Vec<(String, Option<String>)>> {
        if let Some(inst_id) = instrument_id {
            let symbol = inst_id.symbol.inner().as_str();
            let inst_family = extract_inst_family(symbol)?.to_string();
            let exp_time = Self::option_summary_exp_time(symbol)?;
            return Ok(vec![(inst_family, exp_time)]);
        }

        let underlying = Ustr::from(underlying);
        let mut families = AHashSet::new();

        for instrument in self.instruments_cache.load().values() {
            let InstrumentAny::CryptoOption(option) = instrument else {
                continue;
            };

            if option.underlying.code != underlying {
                continue;
            }

            let inst_family = extract_inst_family(option.id.symbol.inner().as_str())?;
            families.insert(inst_family.to_string());
        }

        let mut families: Vec<String> = families.into_iter().collect();
        families.sort_unstable();

        anyhow::ensure!(
            !families.is_empty(),
            "No cached OKX option families for underlying {underlying}; provide a sample instrument or pre-load option instruments"
        );

        Ok(families.into_iter().map(|family| (family, None)).collect())
    }

    fn option_summary_expiry_key(symbol: &str) -> anyhow::Result<String> {
        let parts: Vec<&str> = symbol.split('-').collect();
        anyhow::ensure!(
            parts.len() >= 5,
            "Expected OKX option symbol with expiry, received {symbol}"
        );
        Ok(format!("{}-{}-{}", parts[0], parts[1], parts[2]))
    }

    fn option_summary_exp_time(symbol: &str) -> anyhow::Result<Option<String>> {
        let parts: Vec<&str> = symbol.split('-').collect();
        anyhow::ensure!(
            parts.len() >= 5,
            "Expected OKX option symbol with expiry, received {symbol}"
        );
        Ok(Some(parts[2].to_string()))
    }

    /// Requests the latest index price for the `instrument_id` from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or no index price is returned.
    pub async fn request_index_price(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<IndexPriceUpdate> {
        // Index tickers endpoint requires base pair format (e.g., BTC-USDT)
        let symbol = instrument_id.symbol.inner();
        let (base, quote) = parse_base_quote_from_symbol(symbol.as_str())?;
        let inst_id = format!("{base}-{quote}");

        let mut params = GetIndexTickerParamsBuilder::default();
        params.inst_id(Ustr::from(&inst_id));
        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

        let resp = self
            .inner
            .get_index_tickers(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let raw = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No index price returned from OKX"))?;
        let inst = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let ts_init = self.generate_ts_init();

        let index_price =
            parse_index_price_update(raw, instrument_id, inst.price_precision(), ts_init)
                .map_err(|e| anyhow::anyhow!(e))?;
        Ok(index_price)
    }

    /// Requests an order book snapshot for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or book parsing fails.
    pub async fn request_book_snapshot(
        &self,
        instrument_id: InstrumentId,
        depth: Option<u32>,
    ) -> anyhow::Result<OrderBook> {
        let inst = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let price_precision = inst.price_precision();
        let size_precision = inst.size_precision();

        let params = GetOrderBookParams {
            inst_id: instrument_id.symbol.to_string(),
            sz: depth,
        };

        let resp = self
            .inner
            .get_order_book(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let snapshot = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No order book returned from OKX"))?;

        let ts_event = UnixNanos::from(snapshot.ts * NANOSECONDS_IN_MILLISECOND);
        let mut book = OrderBook::new(instrument_id, BookType::L2_MBP);

        for (i, level) in snapshot.bids.iter().enumerate() {
            let price = parse_price(&level.0, price_precision)?;
            let size = parse_quantity(&level.1, size_precision)?;
            let order = BookOrder::new(OrderSide::Buy, price, size, i as u64);
            book.add(order, 0, i as u64, ts_event);
        }

        let bids_len = snapshot.bids.len();

        for (i, level) in snapshot.asks.iter().enumerate() {
            let price = parse_price(&level.0, price_precision)?;
            let size = parse_quantity(&level.1, size_precision)?;
            let order = BookOrder::new(OrderSide::Sell, price, size, (bids_len + i) as u64);
            book.add(order, 0, (bids_len + i) as u64, ts_event);
        }

        log::info!(
            "Fetched order book for {} with {} bids and {} asks",
            instrument_id,
            snapshot.bids.len(),
            snapshot.asks.len(),
        );

        Ok(book)
    }

    /// Requests an order book snapshot as `OrderBookDeltas` for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    pub async fn request_orderbook_snapshot(
        &self,
        instrument_id: InstrumentId,
        depth: Option<u32>,
    ) -> anyhow::Result<OrderBookDeltas> {
        let inst = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let price_precision = inst.price_precision();
        let size_precision = inst.size_precision();

        let params = GetOrderBookParams {
            inst_id: instrument_id.symbol.to_string(),
            sz: depth,
        };

        let resp = self
            .inner
            .get_order_book(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let snapshot = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No order book returned from OKX"))?;

        let ts_event = UnixNanos::from(snapshot.ts * NANOSECONDS_IN_MILLISECOND);
        let total_levels = snapshot.bids.len() + snapshot.asks.len();
        let mut deltas = Vec::with_capacity(total_levels + 1);

        let mut clear = OrderBookDelta::clear(instrument_id, 0, ts_event, ts_event);

        if total_levels == 0 {
            clear.flags |= RecordFlag::F_LAST as u8;
        }
        deltas.push(clear);

        let mut processed = 0_usize;

        for (i, level) in snapshot.bids.iter().enumerate() {
            let price = parse_price(&level.0, price_precision)?;
            let size = parse_quantity(&level.1, size_precision)?;
            let order = BookOrder::new(OrderSide::Buy, price, size, i as u64);
            processed += 1;
            let mut flags = RecordFlag::F_SNAPSHOT as u8;

            if processed == total_levels {
                flags |= RecordFlag::F_LAST as u8;
            }
            deltas.push(OrderBookDelta::new(
                instrument_id,
                BookAction::Add,
                order,
                flags,
                0,
                ts_event,
                ts_event,
            ));
        }

        let bids_len = snapshot.bids.len();

        for (i, level) in snapshot.asks.iter().enumerate() {
            let price = parse_price(&level.0, price_precision)?;
            let size = parse_quantity(&level.1, size_precision)?;
            let order = BookOrder::new(OrderSide::Sell, price, size, (bids_len + i) as u64);
            processed += 1;
            let mut flags = RecordFlag::F_SNAPSHOT as u8;

            if processed == total_levels {
                flags |= RecordFlag::F_LAST as u8;
            }
            deltas.push(OrderBookDelta::new(
                instrument_id,
                BookAction::Add,
                order,
                flags,
                0,
                ts_event,
                ts_event,
            ));
        }

        log::info!(
            "Fetched order book snapshot for {} with {} bids and {} asks",
            instrument_id,
            snapshot.bids.len(),
            snapshot.asks.len(),
        );

        OrderBookDeltas::new_checked(instrument_id, deltas)
            .context("failed to assemble OrderBookDeltas from OKX snapshot")
    }

    /// Requests historical funding rates for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or parsing fails.
    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>> {
        let mut params = GetFundingRateHistoryParams {
            inst_id: instrument_id.symbol.to_string(),
            ..Default::default()
        };

        // OKX uses "before" for newer-than and "after" for older-than
        if let Some(start) = start {
            params.before = Some(start.timestamp_millis().to_string());
        }

        if let Some(end) = end {
            params.after = Some(end.timestamp_millis().to_string());
        }

        params.limit = limit;

        let resp = self
            .inner
            .get_funding_rate_history(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let mut rates = Vec::with_capacity(resp.len());

        for window in resp.windows(2) {
            let raw = &window[0];
            let interval_millis = raw
                .funding_time
                .checked_sub(window[1].funding_time)
                .context("funding interval negative, funding rates out of order")?;
            let rate = parse_funding_rate(raw, instrument_id, Some(interval_millis))?;
            rates.push(rate);
        }

        if let Some(last_raw) = resp.last() {
            // oldest funding update has no previous one to compute interval
            let rate = parse_funding_rate(last_raw, instrument_id, None)?;
            rates.push(rate);
        }

        // OKX returns newest-first; reverse to chronological order so that
        // cache.add_funding_rates (which push_fronts) leaves the newest at front
        rates.reverse();

        log::info!(
            "Fetched {} funding rates for {}",
            rates.len(),
            instrument_id,
        );

        Ok(rates)
    }

    /// Requests trades for the `instrument_id` and `start` -> `end` time range.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or trade 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>> {
        const OKX_TRADES_MAX_LIMIT: u32 = 100;
        const MAX_PAGES: usize = 500;
        const MAX_CONSECUTIVE_EMPTY: usize = 3;

        #[derive(Clone, Copy, Debug, PartialEq, Eq)]
        enum Mode {
            Latest,
            Backward,
            Range,
        }

        let limit = if limit == Some(0) { None } else { limit };

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

        let now = Utc::now();

        if let Some(s) = start
            && s > now
        {
            return Ok(Vec::new());
        }

        let end = if let Some(e) = end
            && e > now
        {
            Some(now)
        } else {
            end
        };

        let mode = match (start, end) {
            (None, None) => Mode::Latest,
            (Some(_), None) => Mode::Backward,
            (None, Some(_)) => Mode::Backward,
            (Some(_), Some(_)) => Mode::Range,
        };

        let start_ms = start.map(|s| s.timestamp_millis());
        let end_ms = end.map(|e| e.timestamp_millis());

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

        // Historical pagination walks backwards using trade IDs, OKX does not honour timestamps for
        // standalone `before` requests (type=2)
        if matches!(mode, Mode::Backward | Mode::Range) {
            let mut before_trade_id: Option<String> = None;
            let mut pages = 0usize;
            let mut page_results: Vec<Vec<TradeTick>> = Vec::new();
            let mut seen_trades: AHashSet<(String, i64)> = AHashSet::new();
            let mut unique_count = 0usize;
            let mut consecutive_empty_pages = 0usize;

            // Only apply default limit when there's no start boundary
            // (start provides a natural stopping point, end alone allows infinite backward pagination)
            let effective_limit = if start.is_some() {
                limit.unwrap_or(u32::MAX)
            } else {
                limit.unwrap_or(OKX_TRADES_MAX_LIMIT)
            };

            log::debug!(
                "Starting trades pagination: mode={mode:?}, start={start:?}, end={end:?}, limit={limit:?}, effective_limit={effective_limit}"
            );

            loop {
                if pages >= MAX_PAGES {
                    log::warn!("Hit MAX_PAGES limit of {MAX_PAGES}");
                    break;
                }

                if effective_limit < u32::MAX && unique_count >= effective_limit as usize {
                    log::debug!("Reached effective limit: unique_count={unique_count}");
                    break;
                }

                let remaining = (effective_limit as usize).saturating_sub(unique_count);
                let page_cap = remaining.min(OKX_TRADES_MAX_LIMIT as usize) as u32;

                log::debug!(
                    "Requesting page {}: before_id={:?}, page_cap={}, unique_count={}",
                    pages + 1,
                    before_trade_id,
                    page_cap,
                    unique_count
                );

                let mut params_builder = GetTradesParamsBuilder::default();
                params_builder
                    .inst_id(instrument_id.symbol.inner())
                    .limit(page_cap)
                    .pagination_type(1);

                // Use 'after' to get older trades (OKX API: after=cursor means < cursor)
                if let Some(ref before_id) = before_trade_id {
                    params_builder.after(before_id.clone());
                }

                let params = params_builder.build().map_err(anyhow::Error::new)?;
                let raw = self
                    .inner
                    .get_history_trades(params)
                    .await
                    .map_err(anyhow::Error::new)?;

                log::debug!("Received {} raw trades from API", raw.len());

                if let (Some(first), Some(last)) = (raw.first(), raw.last()) {
                    log::debug!(
                        "Raw response trade ID range: first={} (newest), last={} (oldest)",
                        first.trade_id,
                        last.trade_id,
                    );
                }

                if raw.is_empty() {
                    log::debug!("API returned empty page, stopping pagination");
                    break;
                }

                pages += 1;

                let mut page_trades: Vec<TradeTick> = Vec::with_capacity(raw.len());
                let mut hit_start_boundary = false;
                let mut filtered_out = 0usize;
                let mut duplicates = 0usize;

                for r in &raw {
                    match parse_trade_tick(
                        r,
                        instrument_id,
                        inst.price_precision(),
                        inst.size_precision(),
                        ts_init,
                    ) {
                        Ok(trade) => {
                            let ts_ms = trade.ts_event.as_i64() / 1_000_000;

                            if let Some(e_ms) = end_ms
                                && ts_ms > e_ms
                            {
                                filtered_out += 1;
                                continue;
                            }

                            if let Some(s_ms) = start_ms
                                && ts_ms < s_ms
                            {
                                hit_start_boundary = true;
                                filtered_out += 1;
                                break;
                            }

                            let trade_key = (trade.trade_id.to_string(), trade.ts_event.as_i64());
                            if seen_trades.insert(trade_key) {
                                unique_count += 1;
                                page_trades.push(trade);
                            } else {
                                duplicates += 1;
                            }
                        }
                        Err(e) => log::error!("{e}"),
                    }
                }

                log::debug!(
                    "Page {} processed: {} trades kept, {} filtered out, {} duplicates, hit_start_boundary={}",
                    pages,
                    page_trades.len(),
                    filtered_out,
                    duplicates,
                    hit_start_boundary
                );

                // Extract oldest unique trade ID for next page cursor
                let oldest_trade_id = if page_trades.is_empty() {
                    // Only apply consecutive empty guard if we've already collected some trades
                    // This allows historical backfills to paginate through empty prelude
                    if unique_count > 0 {
                        consecutive_empty_pages += 1;
                        if consecutive_empty_pages >= MAX_CONSECUTIVE_EMPTY {
                            log::debug!(
                                "Stopping: {consecutive_empty_pages} consecutive pages with no trades in range after collecting {unique_count} trades"
                            );
                            break;
                        }
                    }
                    // No unique trades on page, use raw response for cursor
                    raw.last().map(|t| {
                        let id = t.trade_id.to_string();
                        log::debug!(
                            "Setting cursor from raw response (no unique trades): oldest_id={id}"
                        );
                        id
                    })
                } else {
                    // Use oldest deduplicated trade ID before reversing
                    let oldest_id = page_trades.last().map(|t| {
                        let id = t.trade_id.to_string();
                        log::debug!(
                            "Setting cursor from deduplicated trades: oldest_id={}, ts_event={}",
                            id,
                            t.ts_event.as_i64()
                        );
                        id
                    });
                    page_trades.reverse();
                    page_results.push(page_trades);
                    consecutive_empty_pages = 0;
                    oldest_id
                };

                if let Some(ref old_id) = before_trade_id
                    && oldest_trade_id.as_ref() == Some(old_id)
                {
                    break;
                }

                if oldest_trade_id.is_none() {
                    break;
                }

                before_trade_id = oldest_trade_id;

                if hit_start_boundary {
                    break;
                }

                tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
            }

            log::debug!(
                "Pagination complete: {pages} pages, {unique_count} unique trades collected"
            );

            let mut out: Vec<TradeTick> = Vec::new();

            for page in page_results.into_iter().rev() {
                out.extend(page);
            }

            // Deduplicate by (trade_id, ts_event) composite key
            let mut dedup_keys = AHashSet::new();
            let pre_dedup_len = out.len();
            out.retain(|trade| {
                dedup_keys.insert((trade.trade_id.to_string(), trade.ts_event.as_i64()))
            });

            if out.len() < pre_dedup_len {
                log::debug!(
                    "Removed {} duplicate trades during final dedup",
                    pre_dedup_len - out.len()
                );
            }

            if let Some(lim) = limit
                && lim > 0
                && out.len() > lim as usize
            {
                let excess = out.len() - lim as usize;
                log::debug!("Trimming {excess} oldest trades to respect limit={lim}");
                out.drain(0..excess);
            }

            log::debug!("Returning {} trades", out.len());
            return Ok(out);
        }

        let req_limit = limit
            .unwrap_or(OKX_TRADES_MAX_LIMIT)
            .min(OKX_TRADES_MAX_LIMIT);
        let params = GetTradesParamsBuilder::default()
            .inst_id(instrument_id.symbol.inner())
            .limit(req_limit)
            .build()
            .map_err(anyhow::Error::new)?;

        let raw = self
            .inner
            .get_history_trades(params)
            .await
            .map_err(anyhow::Error::new)?;

        let mut trades: Vec<TradeTick> = Vec::with_capacity(raw.len());

        for r in &raw {
            match parse_trade_tick(
                r,
                instrument_id,
                inst.price_precision(),
                inst.size_precision(),
                ts_init,
            ) {
                Ok(trade) => trades.push(trade),
                Err(e) => log::error!("{e}"),
            }
        }

        // OKX returns newest-first, reverse to oldest-first
        trades.reverse();

        if let Some(lim) = limit
            && lim > 0
            && trades.len() > lim as usize
        {
            trades.drain(0..trades.len() - lim as usize);
        }

        Ok(trades)
    }

    /// Requests historical bars for the given bar type and time range.
    ///
    /// The aggregation source must be `EXTERNAL`. Time range validation ensures start < end.
    /// Returns bars sorted oldest to newest.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # Endpoint Selection
    ///
    /// The OKX API has different endpoints with different limits:
    /// - Regular endpoint (`/api/v5/market/candles`): ≤ 300 rows/call, ≤ 40 req/2s
    ///   - Used when: start is None OR age ≤ 100 days
    /// - History endpoint (`/api/v5/market/history-candles`): ≤ 100 rows/call, ≤ 20 req/2s
    ///   - Used when: start is Some AND age > 100 days
    ///
    /// Age is calculated as `Utc::now() - start` at the time of the first request.
    ///
    /// # Supported Aggregations
    ///
    /// Maps to OKX bar query parameter:
    /// - `Second` → `{n}s`
    /// - `Minute` → `{n}m`
    /// - `Hour` → `{n}H`
    /// - `Day` → `{n}D`
    /// - `Week` → `{n}W`
    /// - `Month` → `{n}M`
    ///
    /// # Pagination
    ///
    /// - Uses `before` parameter for backwards pagination
    /// - Pages backwards from end time (or now) to start time
    /// - Stops when: limit reached, time window covered, or API returns empty
    /// - Rate limit safety: ≥ 50ms between requests
    ///
    /// # References
    ///
    /// - <https://tr.okx.com/docs-v5/en/#order-book-trading-market-data-get-candlesticks>
    /// - <https://tr.okx.com/docs-v5/en/#order-book-trading-market-data-get-candlesticks-history>
    pub async fn request_bars(
        &self,
        bar_type: BarType,
        start: Option<DateTime<Utc>>,
        mut end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<Bar>> {
        const HISTORY_SPLIT_DAYS: i64 = 100;
        const MAX_PAGES_SOFT: usize = 500;

        #[derive(Clone, Copy, Debug, PartialEq, Eq)]
        enum Mode {
            Latest,
            Backward,
            Range,
        }

        let limit = if limit == Some(0) { None } else { limit };

        anyhow::ensure!(
            bar_type.aggregation_source() == AggregationSource::External,
            "Only EXTERNAL aggregation is supported"
        );

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

        let now = Utc::now();

        if let Some(s) = start
            && s > now
        {
            return Ok(Vec::new());
        }

        if let Some(e) = end
            && e > now
        {
            end = Some(now);
        }

        let spec = bar_type.spec();
        let step = spec.step.get();
        let bar_param = match spec.aggregation {
            BarAggregation::Second => format!("{step}s"),
            BarAggregation::Minute => format!("{step}m"),
            BarAggregation::Hour => format!("{step}H"),
            BarAggregation::Day => format!("{step}D"),
            BarAggregation::Week => format!("{step}W"),
            BarAggregation::Month => format!("{step}M"),
            a => anyhow::bail!("OKX does not support {a:?} aggregation"),
        };

        let slot_ms: i64 = match spec.aggregation {
            BarAggregation::Second => (step as i64) * 1_000,
            BarAggregation::Minute => (step as i64) * 60_000,
            BarAggregation::Hour => (step as i64) * 3_600_000,
            BarAggregation::Day => (step as i64) * 86_400_000,
            BarAggregation::Week => (step as i64) * 7 * 86_400_000,
            BarAggregation::Month => (step as i64) * 30 * 86_400_000,
            _ => unreachable!("Unsupported aggregation should have been caught above"),
        };
        let slot_ns: i64 = slot_ms * 1_000_000;

        let mode = match (start, end) {
            (None, None) => Mode::Latest,
            (Some(_), None) => Mode::Backward, // Changed: when only start is provided, work backward from now
            (None, Some(_)) => Mode::Backward,
            (Some(_), Some(_)) => Mode::Range,
        };

        let start_ns = start.and_then(|s| s.timestamp_nanos_opt());
        let end_ns = end.and_then(|e| e.timestamp_nanos_opt());

        // Floor start and ceiling end to bar boundaries for cleaner API requests
        let start_ms = start.map(|s| {
            let ms = s.timestamp_millis();

            if slot_ms > 0 {
                (ms / slot_ms) * slot_ms // Floor to nearest bar boundary
            } else {
                ms
            }
        });
        let end_ms = end.map(|e| {
            let ms = e.timestamp_millis();

            if slot_ms > 0 {
                ((ms + slot_ms - 1) / slot_ms) * slot_ms // Ceiling to nearest bar boundary
            } else {
                ms
            }
        });
        let now_ms = now.timestamp_millis();

        let symbol = bar_type.instrument_id().symbol;
        let inst = self.instrument_from_cache(symbol.inner())?;

        let mut out: Vec<Bar> = Vec::new();
        let mut pages = 0usize;

        // IMPORTANT: OKX API has COUNTER-INTUITIVE semantics (same for bars and trades):
        // - after=X returns records with timestamp < X (upper bound, despite the name!)
        // - before=X returns records with timestamp > X (lower bound, despite the name!)
        // For Range [start, end], use: before=start (lower bound), after=end (upper bound)
        let mut after_ms: Option<i64> = match mode {
            Mode::Range => end_ms.or(Some(now_ms)), // Upper bound: bars < end
            _ => None,
        };
        let mut before_ms: Option<i64> = match mode {
            Mode::Backward => end_ms.map(|v| v.saturating_sub(1)),
            Mode::Range => start_ms, // Lower bound: bars > start
            Mode::Latest => None,
        };

        // For Range mode, we'll paginate backwards like Backward mode
        let mut forward_prepend_mode = matches!(mode, Mode::Range);

        // Adjust before_ms to ensure we get data from the API
        // OKX API might not have bars for the very recent past
        // This handles both explicit end=now and the actor layer setting end=now when it's None
        if matches!(mode, Mode::Backward | Mode::Range)
            && let Some(b) = before_ms
        {
            // OKX endpoints have different data availability windows:
            // - Regular endpoint: has most recent data but limited depth
            // - History endpoint: has deep history but lags behind current time
            // Use a small buffer to avoid the "dead zone"
            let buffer_ms = slot_ms.max(60_000); // At least 1 minute or 1 bar
            if b >= now_ms.saturating_sub(buffer_ms) {
                before_ms = Some(now_ms.saturating_sub(buffer_ms));
            }
        }

        let mut have_latest_first_page = false;
        let mut progressless_loops = 0u8;

        loop {
            if let Some(lim) = limit
                && lim > 0
                && out.len() >= lim as usize
            {
                break;
            }

            if pages >= MAX_PAGES_SOFT {
                break;
            }

            let pivot_ms = if let Some(a) = after_ms {
                a
            } else if let Some(b) = before_ms {
                b
            } else {
                now_ms
            };
            // Choose endpoint based on how old the data is:
            // - Use regular endpoint for recent data (< 1 hour old)
            // - Use history endpoint for older data (> 1 hour old)
            // This avoids the "gap" where history endpoint has no recent data
            // and regular endpoint has limited depth
            let age_ms = now_ms.saturating_sub(pivot_ms);
            let age_hours = age_ms / (60 * 60 * 1000);
            let using_history = age_hours > 1; // Use history if data is > 1 hour old

            let page_ceiling = if using_history { 100 } else { 300 };
            let remaining = limit
                .filter(|&l| l > 0) // Treat limit=0 as no limit
                .map_or(page_ceiling, |l| (l as usize).saturating_sub(out.len()));
            let page_cap = remaining.min(page_ceiling);

            let mut p = GetCandlesticksParamsBuilder::default();
            p.inst_id(symbol.as_str())
                .bar(&bar_param)
                .limit(page_cap as u32);

            // Track whether this planned request uses BEFORE or AFTER.
            let mut req_used_before = false;

            match mode {
                Mode::Latest => {
                    if have_latest_first_page && let Some(b) = before_ms {
                        p.before_ms(b);
                        req_used_before = true;
                    }
                }
                Mode::Backward => {
                    // Use 'after' to get older bars (OKX API: after=cursor means < cursor)
                    if let Some(b) = before_ms {
                        p.after_ms(b);
                    }
                }
                Mode::Range => {
                    // For Range mode, use both after and before to specify the full range
                    // This is much more efficient than pagination
                    if let Some(a) = after_ms {
                        p.after_ms(a);
                    }

                    if let Some(b) = before_ms {
                        p.before_ms(b);
                        req_used_before = true;
                    }
                }
            }

            let params = p.build().map_err(anyhow::Error::new)?;

            let mut raw = if using_history {
                self.inner
                    .get_history_candles(params.clone())
                    .await
                    .map_err(anyhow::Error::new)?
            } else {
                self.inner
                    .get_candles(params.clone())
                    .await
                    .map_err(anyhow::Error::new)?
            };

            // --- Fallbacks on empty page ---
            if raw.is_empty() {
                // LATEST: retry same cursor via history, then step back a page-interval before giving up
                if matches!(mode, Mode::Latest)
                    && have_latest_first_page
                    && !using_history
                    && let Some(b) = before_ms
                {
                    let mut p2 = GetCandlesticksParamsBuilder::default();
                    p2.inst_id(symbol.as_str())
                        .bar(&bar_param)
                        .limit(page_cap as u32);
                    p2.before_ms(b);
                    let params2 = p2.build().map_err(anyhow::Error::new)?;
                    let raw2 = self
                        .inner
                        .get_history_candles(params2)
                        .await
                        .map_err(anyhow::Error::new)?;

                    if raw2.is_empty() {
                        // Step back one page interval and retry loop
                        let jump = (page_cap as i64).saturating_mul(slot_ms.max(1));
                        before_ms = Some(b.saturating_sub(jump));
                        progressless_loops = progressless_loops.saturating_add(1);
                        if progressless_loops >= 3 {
                            break;
                        }
                        continue;
                    } else {
                        raw = raw2;
                    }
                }

                // Range mode doesn't need special bootstrap - it uses the normal flow with before_ms set

                // If still empty: for Range after first page, try a single backstep window using BEFORE
                if raw.is_empty() && matches!(mode, Mode::Range) && pages > 0 {
                    let backstep_ms = (page_cap as i64).saturating_mul(slot_ms.max(1));
                    let pivot_back = after_ms.unwrap_or(now_ms).saturating_sub(backstep_ms);

                    let mut p2 = GetCandlesticksParamsBuilder::default();
                    p2.inst_id(symbol.as_str())
                        .bar(&bar_param)
                        .limit(page_cap as u32)
                        .before_ms(pivot_back);
                    let params2 = p2.build().map_err(anyhow::Error::new)?;
                    let raw2 = if (now_ms.saturating_sub(pivot_back)) / (24 * 60 * 60 * 1000)
                        > HISTORY_SPLIT_DAYS
                    {
                        self.inner.get_history_candles(params2).await
                    } else {
                        self.inner.get_candles(params2).await
                    }
                    .map_err(anyhow::Error::new)?;

                    if raw2.is_empty() {
                        break;
                    } else {
                        raw = raw2;
                        forward_prepend_mode = true;
                        req_used_before = true;
                    }
                }

                // First LATEST page empty: jump back >100d to force history, then continue loop
                if raw.is_empty()
                    && matches!(mode, Mode::Latest)
                    && !have_latest_first_page
                    && !using_history
                {
                    let jump_days_ms = (HISTORY_SPLIT_DAYS + 1) * 86_400_000;
                    before_ms = Some(now_ms.saturating_sub(jump_days_ms));
                    have_latest_first_page = true;
                    continue;
                }

                // Still empty for any other case? Just break.
                if raw.is_empty() {
                    break;
                }
            }
            // --- end fallbacks ---

            pages += 1;

            // Parse, oldest → newest
            let ts_init = self.generate_ts_init();
            let mut page: Vec<Bar> = Vec::with_capacity(raw.len());

            for r in &raw {
                page.push(parse_candlestick(
                    r,
                    bar_type,
                    inst.price_precision(),
                    inst.size_precision(),
                    ts_init,
                )?);
            }
            page.reverse();

            let page_oldest_ms = page.first().map(|b| b.ts_event.as_i64() / 1_000_000);
            let page_newest_ms = page.last().map(|b| b.ts_event.as_i64() / 1_000_000);

            // Range filter (inclusive)
            // For Range mode, if we have no bars yet and this is an early page,
            // be more tolerant with the start boundary to handle gaps in data
            let mut filtered: Vec<Bar> = if matches!(mode, Mode::Range)
                && out.is_empty()
                && pages < 2
            {
                // On first pages of Range mode with no data yet, include the most recent bar
                // even if it's slightly before our start time (within 2 bar periods)
                // BUT we want ALL bars in the page that are within our range
                let tolerance_ns = slot_ns * 2; // Allow up to 2 bar periods before start

                // Debug: log the page range
                if let (Some(first), Some(last)) = (page.first(), page.last()) {
                    log::debug!(
                        "Range mode bootstrap page: {} bars from {} to {}, filtering with start={:?} end={:?}",
                        page.len(),
                        first.ts_event.as_i64() / 1_000_000,
                        last.ts_event.as_i64() / 1_000_000,
                        start_ms,
                        end_ms,
                    );
                }

                let result: Vec<Bar> = page
                    .clone()
                    .into_iter()
                    .filter(|b| {
                        let ts = b.ts_event.as_i64();
                        // Accept bars from (start - tolerance) to end
                        let ok_after =
                            start_ns.is_none_or(|sns| ts >= sns.saturating_sub(tolerance_ns));
                        let ok_before = end_ns.is_none_or(|ens| ts <= ens);
                        ok_after && ok_before
                    })
                    .collect();

                result
            } else {
                // Normal filtering
                page.clone()
                    .into_iter()
                    .filter(|b| {
                        let ts = b.ts_event.as_i64();
                        let ok_after = start_ns.is_none_or(|sns| ts >= sns);
                        let ok_before = end_ns.is_none_or(|ens| ts <= ens);
                        ok_after && ok_before
                    })
                    .collect()
            };

            if !page.is_empty() && filtered.is_empty() {
                // For Range mode, if all bars are before our start time, there's no point continuing
                if matches!(mode, Mode::Range)
                    && !forward_prepend_mode
                    && let (Some(newest_ms), Some(start_ms)) = (page_newest_ms, start_ms)
                    && newest_ms < start_ms.saturating_sub(slot_ms * 2)
                {
                    // Bars are too old (more than 2 bar periods before start), stop
                    break;
                }
            }

            // Track contribution for progress guard
            let contribution;

            if out.is_empty() {
                contribution = filtered.len();
                out = filtered;
            } else {
                match mode {
                    Mode::Backward | Mode::Latest => {
                        if let Some(first) = out.first() {
                            filtered.retain(|b| b.ts_event < first.ts_event);
                        }
                        contribution = filtered.len();
                        if contribution != 0 {
                            let mut new_out = Vec::with_capacity(out.len() + filtered.len());
                            new_out.extend_from_slice(&filtered);
                            new_out.extend_from_slice(&out);
                            out = new_out;
                        }
                    }
                    Mode::Range => {
                        if forward_prepend_mode || req_used_before {
                            // We are backfilling older pages: prepend them.
                            if let Some(first) = out.first() {
                                filtered.retain(|b| b.ts_event < first.ts_event);
                            }
                            contribution = filtered.len();
                            if contribution != 0 {
                                let mut new_out = Vec::with_capacity(out.len() + filtered.len());
                                new_out.extend_from_slice(&filtered);
                                new_out.extend_from_slice(&out);
                                out = new_out;
                            }
                        } else {
                            // Normal forward: append newer pages.
                            if let Some(last) = out.last() {
                                filtered.retain(|b| b.ts_event > last.ts_event);
                            }
                            contribution = filtered.len();
                            out.extend(filtered);
                        }
                    }
                }
            }

            // Duplicate-window mitigation for Latest/Backward/Range
            if contribution == 0
                && matches!(mode, Mode::Latest | Mode::Backward | Mode::Range)
                && let Some(b) = before_ms
            {
                let jump = (page_cap as i64).saturating_mul(slot_ms.max(1));
                let new_b = b.saturating_sub(jump);
                if new_b != b {
                    before_ms = Some(new_b);
                }
            }

            if contribution == 0 {
                progressless_loops = progressless_loops.saturating_add(1);
                if progressless_loops >= 3 {
                    break;
                }
            } else {
                progressless_loops = 0;

                // Advance cursors only when we made progress
                match mode {
                    Mode::Latest | Mode::Backward => {
                        if let Some(oldest) = page_oldest_ms {
                            before_ms = Some(oldest.saturating_sub(1));
                            have_latest_first_page = true;
                        } else {
                            break;
                        }
                    }
                    Mode::Range => {
                        if forward_prepend_mode || req_used_before {
                            if let Some(oldest) = page_oldest_ms {
                                // Move back by at least one bar period to avoid getting the same data
                                let jump_back = slot_ms.max(60_000); // At least 1 minute
                                before_ms = Some(oldest.saturating_sub(jump_back));
                                after_ms = None;
                            } else {
                                break;
                            }
                        } else if let Some(newest) = page_newest_ms {
                            after_ms = Some(newest.saturating_add(1));
                            before_ms = None;
                        } else {
                            break;
                        }
                    }
                }
            }

            // Stop conditions
            if let Some(lim) = limit
                && lim > 0
                && out.len() >= lim as usize
            {
                break;
            }

            if let Some(ens) = end_ns
                && let Some(last) = out.last()
                && last.ts_event.as_i64() >= ens
            {
                break;
            }

            if let Some(sns) = start_ns
                && let Some(first) = out.first()
                && (matches!(mode, Mode::Backward) || forward_prepend_mode)
                && first.ts_event.as_i64() <= sns
            {
                // For Range mode, check if we have all bars up to the end time
                if matches!(mode, Mode::Range) {
                    // Don't stop if we haven't reached the end time yet
                    if let Some(ens) = end_ns
                        && let Some(last) = out.last()
                    {
                        let last_ts = last.ts_event.as_i64();
                        if last_ts < ens {
                            // We have bars before start but haven't reached end, need to continue forward
                            // Switch from backward to forward pagination
                            forward_prepend_mode = false;
                            after_ms = Some((last_ts / 1_000_000).saturating_add(1));
                            before_ms = None;
                            continue;
                        }
                    }
                }
                break;
            }

            tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
        }

        // Final rescue for FORWARD/RANGE when nothing gathered
        if out.is_empty() && matches!(mode, Mode::Range) {
            let pivot = end_ms.unwrap_or(now_ms.saturating_sub(1));
            let hist = (now_ms.saturating_sub(pivot)) / (24 * 60 * 60 * 1000) > HISTORY_SPLIT_DAYS;
            let mut p = GetCandlesticksParamsBuilder::default();
            p.inst_id(symbol.as_str())
                .bar(&bar_param)
                .limit(300)
                .before_ms(pivot);
            let params = p.build().map_err(anyhow::Error::new)?;
            let raw = if hist {
                self.inner.get_history_candles(params).await
            } else {
                self.inner.get_candles(params).await
            }
            .map_err(anyhow::Error::new)?;

            if !raw.is_empty() {
                let ts_init = self.generate_ts_init();
                let mut page: Vec<Bar> = Vec::with_capacity(raw.len());

                for r in &raw {
                    page.push(parse_candlestick(
                        r,
                        bar_type,
                        inst.price_precision(),
                        inst.size_precision(),
                        ts_init,
                    )?);
                }
                page.reverse();
                out = page
                    .into_iter()
                    .filter(|b| {
                        let ts = b.ts_event.as_i64();
                        let ok_after = start_ns.is_none_or(|sns| ts >= sns);
                        let ok_before = end_ns.is_none_or(|ens| ts <= ens);
                        ok_after && ok_before
                    })
                    .collect();
            }
        }

        // Trim against end bound if needed (keep ≤ end)
        if let Some(ens) = end_ns {
            while out.last().is_some_and(|b| b.ts_event.as_i64() > ens) {
                out.pop();
            }
        }

        // Clamp first bar for Range when using forward pagination
        if matches!(mode, Mode::Range)
            && !forward_prepend_mode
            && let Some(sns) = start_ns
        {
            let lower = sns.saturating_sub(slot_ns);
            while out.first().is_some_and(|b| b.ts_event.as_i64() < lower) {
                out.remove(0);
            }
        }

        // Keep the most recent N bars when limit is specified
        if let Some(lim) = limit
            && lim > 0
            && out.len() > lim as usize
        {
            let start = out.len() - lim as usize;
            out.drain(..start);
        }

        Ok(out)
    }

    /// Requests historical order status reports for the given parameters.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// - <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-order-history-last-7-days>.
    /// - <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-order-history-last-3-months>.
    #[expect(clippy::too_many_arguments)]
    pub async fn request_order_status_reports(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        open_only: bool,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let instrument_type = if let Some(instrument_type) = instrument_type {
            instrument_type
        } else {
            let instrument_id = instrument_id.ok_or_else(|| {
                anyhow::anyhow!("Instrument ID required if `instrument_type` not provided")
            })?;
            let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
            okx_instrument_type(&instrument)?
        };

        let mut history_base = GetOrderHistoryParamsBuilder::default();
        history_base.inst_type(instrument_type);

        if let Some(instrument_id) = instrument_id.as_ref() {
            history_base.inst_id(instrument_id.symbol.inner().to_string());
        }
        let history_base = history_base.build().map_err(|e| anyhow::anyhow!(e))?;

        let mut pending_base = GetOrderListParamsBuilder::default();
        pending_base.inst_type(instrument_type);

        if let Some(instrument_id) = instrument_id.as_ref() {
            pending_base.inst_id(instrument_id.symbol.inner().to_string());
        }
        let pending_base = pending_base.build().map_err(|e| anyhow::anyhow!(e))?;

        let combined_resp = if open_only {
            self.paginate_orders_pending(&pending_base, limit).await?
        } else {
            let (history, pending) = tokio::try_join!(
                self.paginate_orders_history(&history_base, limit),
                self.paginate_orders_pending(&pending_base, limit),
            )?;
            let mut combined_resp = history;
            combined_resp.extend(pending);
            combined_resp
        };

        // Prepare time range filter
        let start_ns = start.map(UnixNanos::from);
        let end_ns = end.map(UnixNanos::from);

        let ts_init = self.generate_ts_init();
        let mut reports = Vec::with_capacity(combined_resp.len());

        // Use a seen filter in case pending orders are within the histories "2hr reserve window"
        let mut seen: AHashSet<String> = AHashSet::new();

        for order in combined_resp {
            let seen_key = if !order.cl_ord_id.is_empty() {
                order.cl_ord_id.as_str().to_string()
            } else if let Some(algo_cl_ord_id) = order
                .algo_cl_ord_id
                .as_ref()
                .filter(|value| !value.as_str().is_empty())
            {
                algo_cl_ord_id.as_str().to_string()
            } else if let Some(algo_id) = order
                .algo_id
                .as_ref()
                .filter(|value| !value.as_str().is_empty())
            {
                algo_id.as_str().to_string()
            } else {
                order.ord_id.as_str().to_string()
            };

            if !seen.insert(seen_key) {
                continue; // Reserved pending already reported
            }

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

            let report = match parse_order_status_report(
                &order,
                account_id,
                inst.id(),
                inst.price_precision(),
                inst.size_precision(),
                ts_init,
            ) {
                Ok(report) => report,
                Err(e) => {
                    log::error!("Failed to parse order status report: {e}");
                    continue;
                }
            };

            if let Some(start_ns) = start_ns
                && report.ts_last < start_ns
            {
                continue;
            }

            if let Some(end_ns) = end_ns
                && report.ts_last > end_ns
            {
                continue;
            }

            reports.push(report);
        }

        Ok(reports)
    }

    // Paginates through order history using `ord_id` as the cursor
    async fn paginate_orders_history(
        &self,
        base: &GetOrderHistoryParams,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OKXOrderHistory>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

        for _ in 0..MAX_RECONCILIATION_PAGES {
            let mut params = base.clone();
            params.after = cursor.take();

            let page = self
                .inner
                .get_orders_history(params)
                .await
                .map_err(|e| anyhow::anyhow!(e))?;

            let page_len = page.len();
            cursor = page.last().map(|o| o.ord_id.to_string());
            all.extend(page);

            if page_len < OKX_PAGE_SIZE {
                exhausted = false;
                break;
            }

            if let Some(lim) = limit
                && all.len() >= lim as usize
            {
                exhausted = false;
                break;
            }
        }

        if exhausted && !all.is_empty() {
            log::warn!(
                "Order history pagination hit {MAX_RECONCILIATION_PAGES} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        if let Some(lim) = limit {
            all.truncate(lim as usize);
        }

        Ok(all)
    }

    // Paginates through pending orders using `ord_id` as the cursor
    async fn paginate_orders_pending(
        &self,
        base: &GetOrderListParams,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OKXOrderHistory>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

        for _ in 0..MAX_RECONCILIATION_PAGES {
            let mut params = base.clone();
            params.after = cursor.take();

            let page = self
                .inner
                .get_orders_pending(params)
                .await
                .map_err(|e| anyhow::anyhow!(e))?;

            let page_len = page.len();
            cursor = page.last().map(|o| o.ord_id.to_string());
            all.extend(page);

            if page_len < OKX_PAGE_SIZE {
                exhausted = false;
                break;
            }

            if let Some(lim) = limit
                && all.len() >= lim as usize
            {
                exhausted = false;
                break;
            }
        }

        if exhausted && !all.is_empty() {
            log::warn!(
                "Pending orders pagination hit {MAX_RECONCILIATION_PAGES} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        if let Some(lim) = limit {
            all.truncate(lim as usize);
        }

        Ok(all)
    }

    // Paginates through transaction details (fills) using `bill_id` as the cursor
    async fn paginate_fills(
        &self,
        base: &GetTransactionDetailsParams,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OKXTransactionDetail>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

        for _ in 0..MAX_RECONCILIATION_PAGES {
            let mut params = base.clone();
            params.after = cursor.take();

            let page = self
                .inner
                .get_fills(params)
                .await
                .map_err(|e| anyhow::anyhow!(e))?;

            let page_len = page.len();
            cursor = page.last().map(|o| o.bill_id.to_string());
            all.extend(page);

            if page_len < OKX_PAGE_SIZE {
                exhausted = false;
                break;
            }

            if let Some(lim) = limit
                && all.len() >= lim as usize
            {
                exhausted = false;
                break;
            }
        }

        if exhausted && !all.is_empty() {
            log::warn!(
                "Fill pagination hit {MAX_RECONCILIATION_PAGES} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        if let Some(lim) = limit {
            all.truncate(lim as usize);
        }

        Ok(all)
    }

    // Paginates through pending algo orders using `algo_id` as the cursor
    async fn paginate_algo_pending(
        &self,
        base: &GetAlgoOrdersParams,
        limit: Option<usize>,
    ) -> anyhow::Result<Vec<OKXOrderAlgo>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

        for _ in 0..MAX_RECONCILIATION_PAGES {
            let mut params = base.clone();
            params.after = cursor.take();

            let page = match self.inner.get_order_algo_pending(params).await {
                Ok(result) => result,
                Err(OKXHttpError::UnexpectedStatus { status, .. })
                    if status == StatusCode::NOT_FOUND =>
                {
                    exhausted = false;
                    break;
                }
                Err(e) => return Err(e.into()),
            };

            let page_len = page.len();
            cursor = page.last().map(|o| o.algo_id.clone());
            all.extend(page);

            if page_len < OKX_PAGE_SIZE {
                exhausted = false;
                break;
            }

            if let Some(lim) = limit
                && all.len() >= lim
            {
                exhausted = false;
                break;
            }
        }

        if exhausted && !all.is_empty() {
            log::warn!(
                "Algo pending pagination hit {MAX_RECONCILIATION_PAGES} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        Ok(all)
    }

    // Paginates through historical algo orders using `algo_id` as the cursor
    async fn paginate_algo_history(
        &self,
        base: &GetAlgoOrdersParams,
        limit: Option<usize>,
    ) -> anyhow::Result<Vec<OKXOrderAlgo>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

        for _ in 0..MAX_RECONCILIATION_PAGES {
            let mut params = base.clone();
            params.after = cursor.take();

            let page = match self.inner.get_order_algo_history(params).await {
                Ok(result) => result,
                Err(OKXHttpError::UnexpectedStatus { status, .. })
                    if status == StatusCode::NOT_FOUND =>
                {
                    exhausted = false;
                    break;
                }
                Err(e) => return Err(e.into()),
            };

            let page_len = page.len();
            cursor = page.last().map(|o| o.algo_id.clone());
            all.extend(page);

            if page_len < OKX_PAGE_SIZE {
                exhausted = false;
                break;
            }

            if let Some(lim) = limit
                && all.len() >= lim
            {
                exhausted = false;
                break;
            }
        }

        if exhausted && !all.is_empty() {
            log::warn!(
                "Algo history pagination hit {MAX_RECONCILIATION_PAGES} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        Ok(all)
    }

    /// Requests fill reports (transaction details) for the given parameters.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-get-transaction-details-last-3-days>.
    pub async fn request_fill_reports(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        start: Option<DateTime<Utc>>,
        end: Option<DateTime<Utc>>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<FillReport>> {
        let mut params = GetTransactionDetailsParamsBuilder::default();

        let instrument_type = if let Some(instrument_type) = instrument_type {
            instrument_type
        } else {
            let instrument_id = instrument_id.ok_or_else(|| {
                anyhow::anyhow!("Instrument ID required if `instrument_type` not provided")
            })?;
            let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
            okx_instrument_type(&instrument)?
        };

        params.inst_type(instrument_type);

        if let Some(instrument_id) = instrument_id {
            let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
            let instrument_type = okx_instrument_type(&instrument)?;
            params.inst_type(instrument_type);
            params.inst_id(instrument_id.symbol.inner().to_string());
        }

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

        let resp = self.paginate_fills(&params, limit).await?;

        // Prepare time range filter
        let start_ns = start.map(UnixNanos::from);
        let end_ns = end.map(UnixNanos::from);

        let ts_init = self.generate_ts_init();
        let mut reports = Vec::with_capacity(resp.len());

        for detail in resp {
            // Skip fills with zero or negative quantity (cancelled orders, etc)
            if detail.fill_sz.is_empty() {
                continue;
            }

            if let Ok(qty) = detail.fill_sz.parse::<f64>() {
                if qty <= 0.0 {
                    continue;
                }
            } else {
                // Skip unparsable quantities
                continue;
            }

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

            let report = match parse_fill_report(
                &detail,
                account_id,
                inst.id(),
                inst.price_precision(),
                inst.size_precision(),
                ts_init,
            ) {
                Ok(report) => report,
                Err(e) => {
                    log::error!("Failed to parse fill report: {e}");
                    continue;
                }
            };

            if let Some(start_ns) = start_ns
                && report.ts_event < start_ns
            {
                continue;
            }

            if let Some(end_ns) = end_ns
                && report.ts_event > end_ns
            {
                continue;
            }

            reports.push(report);
        }

        Ok(reports)
    }

    /// Requests current position status reports for the given parameters.
    ///
    /// # Position Modes
    ///
    /// OKX supports two position modes, which affects how position data is returned:
    ///
    /// ## Net Mode (One-way)
    /// - `posSide` field will be `"net"`
    /// - `pos` field uses **signed quantities**:
    ///   - Positive value = Long position
    ///   - Negative value = Short position
    ///   - Zero = Flat/no position
    ///
    /// ## Long/Short Mode (Hedge/Dual-side)
    /// - `posSide` field will be `"long"` or `"short"`
    /// - `pos` field is **always positive** (use `posSide` to determine actual side)
    /// - Allows holding simultaneous long and short positions on the same instrument
    /// - Position IDs are suffixed with `-LONG` or `-SHORT` for uniqueness
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-positions>
    pub async fn request_position_status_reports(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        let mut params = GetPositionsParamsBuilder::default();

        let instrument_type = if let Some(instrument_type) = instrument_type {
            instrument_type
        } else {
            let instrument_id = instrument_id.ok_or_else(|| {
                anyhow::anyhow!("Instrument ID required if `instrument_type` not provided")
            })?;
            let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
            okx_instrument_type(&instrument)?
        };

        params.inst_type(instrument_type);

        instrument_id
            .as_ref()
            .map(|i| params.inst_id(i.symbol.inner()));

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

        let resp = self
            .inner
            .get_positions(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let ts_init = self.generate_ts_init();
        let mut reports = Vec::with_capacity(resp.len());

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

            match parse_position_status_report(
                &position,
                account_id,
                inst.id(),
                inst.size_precision(),
                ts_init,
            ) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    log::error!("Failed to parse position status report: {e}");
                }
            }
        }

        Ok(reports)
    }

    /// Requests spot margin position status reports from account balance.
    ///
    /// Spot margin positions appear in `/api/v5/account/balance` as balance sheet items
    /// with non-zero `liab` (liability) or `spotInUseAmt` fields, rather than in the
    /// positions endpoint. This method fetches the balance and converts any margin
    /// positions into position status reports.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or parsing fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-balance>
    pub async fn request_spot_margin_position_reports(
        &self,
        account_id: AccountId,
    ) -> anyhow::Result<Vec<PositionStatusReport>> {
        let accounts = self
            .inner
            .get_balance()
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let ts_init = self.generate_ts_init();
        let mut reports = Vec::new();

        // Build a base-currency lookup over the cached spot pairs once per
        // call. Restricting to `CurrencyPair` (spot) ensures a derivative
        // sharing the same base (e.g. `BTC-USDT-SWAP`) is never reported as
        // a spot margin position with the wrong instrument id or size
        // precision.
        //
        // When multiple spot pairs share the same base currency, prefer the
        // dominant OKX quote (USDT, then USDC, then USD) so a live
        // `BTC-USDT` margin position stays reported under `BTC-USDT.OKX`
        // rather than being redirected to `BTC-USD.OKX` or any other
        // lexically-earlier pair. Unknown quotes fall back to a stable
        // lexical order by symbol, matching OKX's own listing precedence
        // and keeping the selection deterministic across runs.
        let cache_snapshot = self.instruments_cache.load();
        let mut candidates: Vec<&InstrumentAny> = cache_snapshot
            .values()
            .filter(|inst| matches!(inst, InstrumentAny::CurrencyPair(_)))
            .collect();
        candidates.sort_by(|a, b| {
            let a_sym = a.id().symbol.as_str().to_string();
            let b_sym = b.id().symbol.as_str().to_string();
            spot_quote_priority(&a_sym)
                .cmp(&spot_quote_priority(&b_sym))
                .then_with(|| a_sym.cmp(&b_sym))
        });

        let mut by_base: AHashMap<Ustr, (InstrumentId, u8)> = AHashMap::new();

        for inst in candidates {
            if let Some(base) = inst.base_currency() {
                let base_code = Ustr::from(base.code.as_str());
                by_base
                    .entry(base_code)
                    .or_insert_with(|| (inst.id(), inst.size_precision()));
            }
        }

        for account in accounts {
            for balance in account.details {
                let ccy_str = balance.ccy.as_str();

                let Some((instrument_id, size_precision)) =
                    by_base.get(&Ustr::from(ccy_str)).copied()
                else {
                    log::debug!("Skipping balance for {ccy_str} - no matching instrument in cache");
                    continue;
                };

                match parse_spot_margin_position_from_balance(
                    &balance,
                    account_id,
                    instrument_id,
                    size_precision,
                    ts_init,
                ) {
                    Ok(Some(report)) => reports.push(report),
                    Ok(None) => {} // No margin position for this currency
                    Err(e) => {
                        log::error!(
                            "Failed to parse spot margin position from balance for {ccy_str}: {e}"
                        );
                    }
                }
            }
        }

        Ok(reports)
    }

    /// Places a regular order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-trade-post-place-order>
    pub async fn place_order(
        &self,
        request: OKXPlaceOrderRequest,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        let body =
            serde_json::to_vec(&request).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXPlaceOrderResponse> = self
            .inner
            .send_request::<_, ()>(Method::POST, "/api/v5/trade/order", None, Some(body), true)
            .await?;

        resp.into_iter()
            .next()
            .ok_or_else(|| OKXHttpError::ValidationError("Empty response".to_string()))
    }

    /// Places an algo order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-post-place-algo-order>
    pub async fn place_algo_order(
        &self,
        request: OKXPlaceAlgoOrderRequest,
    ) -> Result<OKXPlaceAlgoOrderResponse, OKXHttpError> {
        let body =
            serde_json::to_vec(&request).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXPlaceAlgoOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/order-algo",
                None,
                Some(body),
                true,
            )
            .await?;

        let item = resp
            .into_iter()
            .next()
            .ok_or_else(|| OKXHttpError::ValidationError("Empty response".to_string()))?;

        if let Some(ref code) = item.s_code
            && code != "0"
        {
            let msg = item.s_msg.clone().unwrap_or_default();
            return Err(OKXHttpError::OkxError {
                error_code: code.clone(),
                message: msg,
            });
        }

        Ok(item)
    }

    /// Cancels an algo order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-post-cancel-algo-order>
    pub async fn cancel_algo_order(
        &self,
        request: OKXCancelAlgoOrderRequest,
    ) -> Result<OKXCancelAlgoOrderResponse, OKXHttpError> {
        // OKX expects an array for cancel-algos endpoint
        // Serialize once to bytes to keep signing and sending identical
        let body =
            serde_json::to_vec(&[request]).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXCancelAlgoOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/cancel-algos",
                None,
                Some(body),
                true,
            )
            .await?;

        let item = resp
            .into_iter()
            .next()
            .ok_or_else(|| OKXHttpError::ValidationError("Empty response".to_string()))?;

        if let Some(ref code) = item.s_code
            && code != "0"
        {
            let msg = item.s_msg.clone().unwrap_or_default();
            return Err(OKXHttpError::OkxError {
                error_code: code.clone(),
                message: msg,
            });
        }

        Ok(item)
    }

    /// Cancels multiple algo orders via HTTP in a single request.
    ///
    /// Items with non-zero `sCode` are logged as warnings but do not
    /// fail the entire batch.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-post-cancel-algo-order>
    pub async fn cancel_algo_orders(
        &self,
        requests: Vec<OKXCancelAlgoOrderRequest>,
    ) -> Result<Vec<OKXCancelAlgoOrderResponse>, OKXHttpError> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let body =
            serde_json::to_vec(&requests).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXCancelAlgoOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/cancel-algos",
                None,
                Some(body),
                true,
            )
            .await?;

        for item in &resp {
            if let Some(ref code) = item.s_code
                && code != "0"
            {
                let msg = item.s_msg.as_deref().unwrap_or("");
                log::warn!(
                    "Algo cancel rejected: algo_id={} sCode={code} sMsg={msg}",
                    item.algo_id
                );
            }
        }

        Ok(resp)
    }

    /// Cancels advance algo orders (trailing stop, iceberg, TWAP) via HTTP.
    ///
    /// These order types cannot use the standard `cancel-algos` endpoint.
    /// Items with non-zero `sCode` are logged as warnings.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-post-cancel-advance-algo-order>
    pub async fn cancel_advance_algo_orders(
        &self,
        requests: Vec<OKXCancelAlgoOrderRequest>,
    ) -> Result<Vec<OKXCancelAlgoOrderResponse>, OKXHttpError> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let body =
            serde_json::to_vec(&requests).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXCancelAlgoOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/cancel-advance-algos",
                None,
                Some(body),
                true,
            )
            .await?;

        for item in &resp {
            if let Some(ref code) = item.s_code
                && code != "0"
            {
                let msg = item.s_msg.as_deref().unwrap_or("");
                log::warn!(
                    "Advance algo cancel rejected: algo_id={} sCode={code} sMsg={msg}",
                    item.algo_id
                );
            }
        }

        Ok(resp)
    }

    /// Amends an algo order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-post-amend-algo-order>
    pub async fn amend_algo_order(
        &self,
        request: OKXAmendAlgoOrderRequest,
    ) -> Result<OKXAmendAlgoOrderResponse, OKXHttpError> {
        let body =
            serde_json::to_vec(&request).map_err(|e| OKXHttpError::JsonError(e.to_string()))?;

        let resp: Vec<OKXAmendAlgoOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/amend-algos",
                None,
                Some(body),
                true,
            )
            .await?;

        resp.into_iter()
            .next()
            .ok_or_else(|| OKXHttpError::ValidationError("Empty response".to_string()))
    }

    /// Amends an algo order using domain types.
    ///
    /// This is a convenience method that accepts Nautilus domain types
    /// and builds the appropriate OKX request structure internally.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    #[expect(clippy::too_many_arguments)]
    pub async fn amend_algo_order_with_domain_types(
        &self,
        instrument_id: InstrumentId,
        algo_id: String,
        new_trigger_price: Option<Price>,
        new_limit_price: Option<Price>,
        new_quantity: Option<Quantity>,
        new_callback_ratio: Option<String>,
        new_callback_spread: Option<String>,
        new_activation_price: Option<Price>,
    ) -> Result<OKXAmendAlgoOrderResponse, OKXHttpError> {
        let request = OKXAmendAlgoOrderRequest {
            inst_id: instrument_id.symbol.as_str().to_string(),
            algo_id,
            algo_cl_ord_id: None,
            new_sz: new_quantity.map(|q| q.to_string()),
            new_trigger_px: new_trigger_price.map(|p| p.to_string()),
            new_order_px: new_limit_price.map(|p| p.to_string()),
            new_callback_ratio,
            new_callback_spread,
            new_active_px: new_activation_price.map(|p| p.to_string()),
        };

        self.amend_algo_order(request).await
    }

    /// Places an algo order using domain types.
    ///
    /// This is a convenience method that accepts Nautilus domain types
    /// and builds the appropriate OKX request structure internally.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    #[expect(clippy::too_many_arguments)]
    pub async fn place_order_with_domain_types(
        &self,
        instrument_id: InstrumentId,
        td_mode: OKXTradeMode,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: Option<TimeInForce>,
        price: Option<Price>,
        post_only: Option<bool>,
        reduce_only: Option<bool>,
        quote_quantity: Option<bool>,
        position_side: Option<PositionSide>,
        attach_algo_ords: Option<Vec<OKXAttachAlgoOrdRequest>>,
        px_usd: Option<String>,
        px_vol: Option<String>,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        if !OKX_SUPPORTED_ORDER_TYPES.contains(&order_type) {
            return Err(OKXHttpError::ValidationError(format!(
                "Unsupported order type: {order_type:?}",
            )));
        }

        if matches!(
            order_type,
            OrderType::StopMarket
                | OrderType::StopLimit
                | OrderType::MarketIfTouched
                | OrderType::LimitIfTouched
                | OrderType::TrailingStopMarket
        ) {
            return Err(OKXHttpError::ValidationError(
                "Conditional order types must use OKX algo order placement".to_string(),
            ));
        }

        if let Some(tif) = time_in_force
            && !OKX_SUPPORTED_TIME_IN_FORCE.contains(&tif)
        {
            return Err(OKXHttpError::ValidationError(format!(
                "Unsupported time in force: {tif:?}",
            )));
        }

        if !matches!(order_side, OrderSide::Buy | OrderSide::Sell) {
            return Err(OKXHttpError::ValidationError(
                "Invalid order side".to_string(),
            ));
        }

        let instrument = self
            .instrument_from_cache(instrument_id.symbol.inner())
            .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;
        let instrument_type = okx_instrument_type(&instrument)
            .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;

        // OKX options only support limit-style orders
        if instrument_type == OKXInstrumentType::Option
            && matches!(order_type, OrderType::Market | OrderType::MarketToLimit)
        {
            return Err(OKXHttpError::ValidationError(
                "Market orders are not supported for OKX options, use Limit orders instead"
                    .to_string(),
            ));
        }

        let side = OKXSide::from(order_side.as_specified());
        let pos_side = position_side.map(Into::into).or({
            if matches!(
                instrument_type,
                OKXInstrumentType::Swap | OKXInstrumentType::Futures | OKXInstrumentType::Option
            ) {
                Some(OKXPositionSide::Net)
            } else {
                None
            }
        });

        let tgt_ccy = if instrument_type == OKXInstrumentType::Spot
            && order_type == OrderType::Market
            && td_mode == OKXTradeMode::Cash
        {
            match quote_quantity {
                Some(true) => Some(OKXTargetCurrency::QuoteCcy),
                Some(false) if order_side == OrderSide::Buy => Some(OKXTargetCurrency::BaseCcy),
                _ => None,
            }
        } else {
            None
        };

        let (ord_type, px) = if post_only.unwrap_or(false) {
            (OKXOrderType::PostOnly, price)
        } else if let Some(tif) = time_in_force {
            match (order_type, tif) {
                (OrderType::Market, TimeInForce::Fok) => {
                    return Err(OKXHttpError::ValidationError(
                        "Market orders with FOK time-in-force are not supported by OKX. Use Limit order with FOK instead.".to_string(),
                    ));
                }
                (OrderType::Market, TimeInForce::Ioc) => {
                    // optimal_limit_ioc only works for SWAP/FUTURES
                    if matches!(
                        instrument_type,
                        OKXInstrumentType::Spot | OKXInstrumentType::Option
                    ) {
                        (OKXOrderType::Market, price)
                    } else {
                        (OKXOrderType::OptimalLimitIoc, price)
                    }
                }
                (OrderType::Limit, TimeInForce::Fok) => {
                    // OKX uses op_fok for options FOK orders
                    if instrument_type == OKXInstrumentType::Option {
                        (OKXOrderType::OpFok, price)
                    } else {
                        (OKXOrderType::Fok, price)
                    }
                }
                (OrderType::Limit, TimeInForce::Ioc) => (OKXOrderType::Ioc, price),
                _ => (OKXOrderType::from(order_type), price),
            }
        } else {
            (OKXOrderType::from(order_type), price)
        };

        // reduceOnly is not applicable to options per OKX docs
        let reduce_only = if instrument_type == OKXInstrumentType::Option {
            None
        } else {
            reduce_only
        };

        // For options: pxUsd/pxVol are mutually exclusive with px
        let (px, px_usd, px_vol) = if px_usd.is_some() {
            (None, px_usd, None)
        } else if px_vol.is_some() {
            (None, None, px_vol)
        } else {
            (px.map(|p| p.to_string()), None, None)
        };

        let request = OKXPlaceOrderRequest {
            inst_id: instrument_id.symbol.as_str().to_string(),
            td_mode,
            ccy: None,
            cl_ord_id: Some(client_order_id.as_str().to_string()),
            tag: Some(OKX_NAUTILUS_BROKER_ID.to_string()),
            side,
            pos_side,
            ord_type,
            sz: quantity.to_string(),
            px,
            px_usd,
            px_vol,
            reduce_only,
            tgt_ccy,
            attach_algo_ords,
        };

        self.place_order(request).await
    }

    /// Places an algo order using domain types.
    ///
    /// This is a convenience method that accepts Nautilus domain types
    /// and builds the appropriate OKX request structure internally.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    #[expect(clippy::too_many_arguments)]
    pub async fn place_algo_order_with_domain_types(
        &self,
        instrument_id: InstrumentId,
        td_mode: OKXTradeMode,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        trigger_price: Option<Price>,
        trigger_type: Option<TriggerType>,
        limit_price: Option<Price>,
        reduce_only: Option<bool>,
        close_fraction: Option<String>,
        callback_ratio: Option<String>,
        callback_spread: Option<String>,
        activation_price: Option<Price>,
    ) -> Result<OKXPlaceAlgoOrderResponse, OKXHttpError> {
        if !matches!(order_side, OrderSide::Buy | OrderSide::Sell) {
            return Err(OKXHttpError::ValidationError(
                "Invalid order side".to_string(),
            ));
        }

        let okx_side = OKXSide::from(order_side.as_specified());

        // Map trigger type to OKX format
        let trigger_px_type_enum = trigger_type.map_or(OKXTriggerType::Last, Into::into);

        let uses_close_fraction = close_fraction.is_some();
        let (
            algo_type,
            sz,
            trigger_px,
            order_px,
            trigger_px_type,
            sl_trigger_px,
            sl_ord_px,
            sl_trigger_px_type,
            tp_trigger_px,
            tp_ord_px,
            tp_trigger_px_type,
            pos_side,
            reduce_only,
        ) = if uses_close_fraction {
            if order_type == OrderType::TrailingStopMarket {
                return Err(OKXHttpError::ValidationError(
                    "OKX close_fraction does not support TrailingStopMarket".to_string(),
                ));
            }

            let trigger_px = trigger_price.map(|p| p.to_string()).ok_or_else(|| {
                OKXHttpError::ValidationError(
                    "OKX close_fraction orders require trigger_price".to_string(),
                )
            })?;

            let close_order_px =
                if matches!(order_type, OrderType::StopLimit | OrderType::LimitIfTouched) {
                    limit_price.map(|p| p.to_string()).ok_or_else(|| {
                        OKXHttpError::ValidationError(format!(
                            "OKX {order_type:?} close_fraction orders require limit_price"
                        ))
                    })?
                } else {
                    "-1".to_string()
                };

            let (
                sl_trigger_px,
                sl_ord_px,
                sl_trigger_px_type,
                tp_trigger_px,
                tp_ord_px,
                tp_trigger_px_type,
            ) = match order_type {
                OrderType::StopMarket | OrderType::StopLimit => (
                    Some(trigger_px),
                    Some(close_order_px),
                    Some(trigger_px_type_enum),
                    None,
                    None,
                    None,
                ),
                OrderType::MarketIfTouched | OrderType::LimitIfTouched => (
                    None,
                    None,
                    None,
                    Some(trigger_px),
                    Some(close_order_px),
                    Some(trigger_px_type_enum),
                ),
                _ => {
                    return Err(OKXHttpError::ValidationError(format!(
                        "OKX close_fraction is only supported for stop/touched conditional orders, received {order_type:?}"
                    )));
                }
            };

            (
                OKXAlgoOrderType::Conditional,
                None,
                None,
                None,
                None,
                sl_trigger_px,
                sl_ord_px,
                sl_trigger_px_type,
                tp_trigger_px,
                tp_ord_px,
                tp_trigger_px_type,
                Some(OKXPositionSide::Net),
                Some(true),
            )
        } else {
            let algo_type = conditional_order_to_algo_type(order_type)
                .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;

            let order_px = if matches!(order_type, OrderType::StopLimit | OrderType::LimitIfTouched)
            {
                limit_price.map(|p| p.to_string())
            } else if order_type == OrderType::TrailingStopMarket {
                None
            } else {
                Some("-1".to_string())
            };

            (
                algo_type,
                Some(quantity.to_string()),
                trigger_price.map(|p| p.to_string()),
                order_px,
                Some(trigger_px_type_enum),
                None,
                None,
                None,
                None,
                None,
                None,
                None,
                reduce_only,
            )
        };

        let request = OKXPlaceAlgoOrderRequest {
            inst_id: instrument_id.symbol.as_str().to_string(),
            inst_id_code: None,
            td_mode,
            side: okx_side,
            ord_type: algo_type,
            sz,
            algo_cl_ord_id: Some(client_order_id.as_str().to_string()),
            trigger_px,
            order_px,
            trigger_px_type,
            sl_trigger_px,
            sl_ord_px,
            sl_trigger_px_type,
            tp_trigger_px,
            tp_ord_px,
            tp_trigger_px_type,
            tgt_ccy: None,
            pos_side,
            close_position: None,
            tag: Some(OKX_NAUTILUS_BROKER_ID.to_string()),
            reduce_only,
            close_fraction,
            callback_ratio,
            callback_spread,
            active_px: activation_price.map(|p| p.to_string()),
        };

        self.place_algo_order(request).await
    }

    /// Cancels an algo order using domain types.
    ///
    /// This is a convenience method that accepts Nautilus domain types
    /// and builds the appropriate OKX request structure internally.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn cancel_algo_order_with_domain_types(
        &self,
        instrument_id: InstrumentId,
        algo_id: String,
    ) -> Result<OKXCancelAlgoOrderResponse, OKXHttpError> {
        let request = OKXCancelAlgoOrderRequest {
            inst_id: instrument_id.symbol.to_string(),
            inst_id_code: None,
            algo_id: Some(algo_id),
            algo_cl_ord_id: None,
        };

        self.cancel_algo_order(request).await
    }

    /// Requests algo order status reports.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    #[expect(clippy::too_many_arguments)]
    pub async fn request_algo_order_status_reports(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        algo_id: Option<String>,
        algo_client_order_id: Option<ClientOrderId>,
        state: Option<OKXOrderStatus>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        let mut instruments_cache: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
        let has_specific_lookup = algo_id.is_some() || algo_client_order_id.is_some();

        let inst_type = if let Some(inst_type) = instrument_type {
            inst_type
        } else if let Some(inst_id) = instrument_id {
            let instrument = self.instrument_from_cache(inst_id.symbol.inner())?;
            let inst_type = okx_instrument_type(&instrument)?;
            instruments_cache.insert(inst_id.symbol.inner(), instrument);
            inst_type
        } else {
            anyhow::bail!("instrument_type or instrument_id required for algo order query")
        };

        let ts_init = self.generate_ts_init();
        let mut reports = Vec::new();
        let mut seen: AHashSet<(String, String)> = AHashSet::new();

        for ord_type in [
            OKXAlgoOrderType::Oco,
            OKXAlgoOrderType::Conditional,
            OKXAlgoOrderType::Trigger,
            OKXAlgoOrderType::MoveOrderStop,
        ] {
            let mut params_builder = GetAlgoOrdersParamsBuilder::default();
            params_builder.inst_type(inst_type);
            params_builder.ord_type(ord_type);

            if let Some(inst_id) = instrument_id {
                params_builder.inst_id(inst_id.symbol.inner().to_string());
            }

            if let Some(algo_id) = algo_id.as_ref() {
                params_builder.algo_id(algo_id.clone());
            }

            if let Some(client_order_id) = algo_client_order_id.as_ref() {
                params_builder.algo_cl_ord_id(client_order_id.as_str().to_string());
            }

            if let Some(state) = state {
                params_builder.state(state);
            }

            let params = params_builder
                .build()
                .map_err(|e| anyhow::anyhow!(format!("Failed to build algo order params: {e}")))?;

            let remaining = limit.map(|l| (l as usize).saturating_sub(reports.len()));
            let pending = self.paginate_algo_pending(&params, remaining).await?;
            self.collect_algo_reports(
                account_id,
                &pending,
                &mut instruments_cache,
                ts_init,
                &mut seen,
                &mut reports,
            )
            .await?;

            if has_specific_lookup && !reports.is_empty() {
                return Ok(reports);
            }

            if let Some(lim) = limit
                && reports.len() >= lim as usize
            {
                reports.truncate(lim as usize);
                return Ok(reports);
            }

            // OKX's `/orders-algo-history` endpoint rejects calls that
            // carry neither a `state` nor an `algoId` / `algoClOrdId`
            // narrowing with code 50015. The reconciliation path wants
            // only currently-live algo orders (those already appear in
            // the pending response above), so skip the history leg when
            // the caller supplied no narrowing. Specific-lookup callers
            // still hit history because `has_specific_lookup` implies
            // `algoId` or `algoClOrdId`, which the endpoint accepts.
            if state.is_some() || has_specific_lookup {
                let remaining = limit.map(|l| (l as usize).saturating_sub(reports.len()));
                let history = self.paginate_algo_history(&params, remaining).await?;
                self.collect_algo_reports(
                    account_id,
                    &history,
                    &mut instruments_cache,
                    ts_init,
                    &mut seen,
                    &mut reports,
                )
                .await?;

                if has_specific_lookup && !reports.is_empty() {
                    return Ok(reports);
                }

                if let Some(lim) = limit
                    && reports.len() >= lim as usize
                {
                    reports.truncate(lim as usize);
                    return Ok(reports);
                }
            }
        }

        Ok(reports)
    }

    /// Requests an algo order status report by client order identifier.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn request_algo_order_status_report(
        &self,
        account_id: AccountId,
        instrument_id: InstrumentId,
        algo_client_order_id: ClientOrderId,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let reports = self
            .request_algo_order_status_reports(
                account_id,
                None,
                Some(instrument_id),
                None,
                Some(algo_client_order_id),
                None,
                Some(50_u32),
            )
            .await?;

        Ok(reports.into_iter().next())
    }

    /// Exposes raw HTTP client for testing purposes
    pub fn raw_client(&self) -> &Arc<OKXRawHttpClient> {
        &self.inner
    }

    async fn collect_algo_reports(
        &self,
        account_id: AccountId,
        orders: &[OKXOrderAlgo],
        instruments_cache: &mut AHashMap<Ustr, InstrumentAny>,
        ts_init: UnixNanos,
        seen: &mut AHashSet<(String, String)>,
        reports: &mut Vec<OrderStatusReport>,
    ) -> anyhow::Result<()> {
        for order in orders {
            let key = (order.algo_id.clone(), order.algo_cl_ord_id.clone());
            if !seen.insert(key) {
                continue;
            }

            let instrument = if let Some(instrument) = instruments_cache.get(&order.inst_id) {
                instrument.clone()
            } else {
                let Ok(instrument) = self.instrument_from_cache(order.inst_id) else {
                    log::debug!(
                        "Skipping algo order report for instrument not in cache: symbol={}",
                        order.inst_id,
                    );
                    continue;
                };
                instruments_cache.insert(order.inst_id, instrument.clone());
                instrument
            };

            match parse_http_algo_order(order, account_id, &instrument, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    log::error!("Failed to parse algo order report: {e}");
                }
            }
        }

        Ok(())
    }
}

fn parse_http_algo_order(
    order: &OKXOrderAlgo,
    account_id: AccountId,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let ord_px = if order.ord_px.is_empty() {
        "-1".to_string()
    } else {
        order.ord_px.clone()
    };

    let reduce_only = if order.reduce_only.is_empty() {
        "false".to_string()
    } else {
        order.reduce_only.clone()
    };

    let msg = OKXAlgoOrderMsg {
        algo_id: order.algo_id.clone(),
        algo_cl_ord_id: order.algo_cl_ord_id.clone(),
        cl_ord_id: order.cl_ord_id.clone(),
        ord_id: order.ord_id.clone(),
        inst_id: order.inst_id,
        inst_type: order.inst_type,
        ord_type: order.ord_type,
        state: order.state,
        side: order.side,
        pos_side: order.pos_side,
        sz: order.sz.clone(),
        trigger_px: order.trigger_px.clone(),
        trigger_px_type: order.trigger_px_type.unwrap_or(OKXTriggerType::None),
        sl_trigger_px: order.sl_trigger_px.clone(),
        sl_ord_px: order.sl_ord_px.clone(),
        sl_trigger_px_type: order.sl_trigger_px_type.unwrap_or(OKXTriggerType::None),
        tp_trigger_px: order.tp_trigger_px.clone(),
        tp_ord_px: order.tp_ord_px.clone(),
        tp_trigger_px_type: order.tp_trigger_px_type.unwrap_or(OKXTriggerType::None),
        ord_px,
        td_mode: order.td_mode,
        lever: order.lever.clone(),
        reduce_only,
        close_fraction: order.close_fraction.clone(),
        actual_px: order.actual_px.clone(),
        actual_sz: order.actual_sz.clone(),
        notional_usd: order.notional_usd.clone(),
        c_time: order.c_time,
        u_time: order.u_time,
        trigger_time: order.trigger_time.clone(),
        tag: order.tag.clone(),
        callback_ratio: order.callback_ratio.clone(),
        callback_spread: order.callback_spread.clone(),
        active_px: order.active_px.clone(),
        ccy: None,
        tgt_ccy: None,
        fee: None,
        fee_ccy: None,
        advance_ord_type: None,
    };

    parse_algo_order_status_report(&msg, instrument, account_id, ts_init)
}