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, Mutex,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use jiff::Timestamp;
use nautilus_common::{cache::InstrumentLookupError, live::dst::time};
use nautilus_core::{
    AtomicMap, AtomicTime, UnixNanos, 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,
        bar::{get_bar_interval_ns, get_time_bar_bounds},
    },
    enums::{
        AccountType, AggregationSource, BarAggregation, BookAction, BookType, OrderSide,
        OrderStatus, OrderType, PositionSide, RecordFlag, TimeInForce, TriggerType,
    },
    events::AccountState,
    identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
    instruments::{Instrument, InstrumentAny},
    orderbook::OrderBook,
    reports::{FillReport, OrderStatusReport, PositionStatusReport},
    types::{Price, Quantity},
};
use nautilus_network::{
    http::{
        HttpClient, HttpRedirectPolicy, Method, StatusCode, create_standard_nautilus_headers,
        parse_retry_after,
    },
    ratelimiter::quota::Quota,
    retry::{RetryConfig, RetryError, RetryManager},
};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use thiserror::Error;
use tokio_util::sync::CancellationToken;
use ustr::Ustr;

use super::{
    error::OKXHttpError,
    models::{
        OKXAccount, OKXAccountBill, OKXAccountConfiguration, OKXAmendAlgoOrderRequest,
        OKXAmendAlgoOrderResponse, OKXAmendOrderRequest, OKXAttachAlgoOrdRequest,
        OKXCancelAlgoOrderRequest, OKXCancelAlgoOrderResponse, OKXCancelAllSpreadOrdersRequest,
        OKXCancelOrderRequest, OKXCancelOrderResponse, OKXCancelSpreadOrderRequest,
        OKXEventContractEvent, OKXEventContractMarket, OKXEventContractSeries, OKXFeeRate,
        OKXFundingRateHistory, OKXIndexTicker, OKXMarkPrice, OKXOptionSummary, OKXOrderAlgo,
        OKXOrderBookSnapshot, OKXOrderHistory, OKXPlaceAlgoOrderRequest, OKXPlaceAlgoOrderResponse,
        OKXPlaceOrderRequest, OKXPlaceOrderResponse, OKXPlaceSpreadOrderRequest, OKXPosition,
        OKXPositionHistory, OKXPositionTier, OKXPriceLimit, OKXRpiOrderBookSnapshot, OKXServerTime,
        OKXSpread, OKXSpreadOrder, OKXSpreadTrade, OKXTransactionDetail,
    },
    query::{
        ActivateFeatureParams, GetAccountBillsParams, GetAlgoOrderParams,
        GetAlgoOrderParamsBuilder, GetAlgoOrdersParams, GetAlgoOrdersParamsBuilder,
        GetCandlesticksParams, GetCandlesticksParamsBuilder, GetEventContractEventsParams,
        GetEventContractMarketsParams, GetEventContractSeriesParams, GetFundingRateHistoryParams,
        GetIndexTickerParams, GetIndexTickerParamsBuilder, GetInstrumentsParams,
        GetInstrumentsParamsBuilder, GetMarkPriceParams, GetMarkPriceParamsBuilder,
        GetOptionSummaryParams, GetOrderBookParams, GetOrderHistoryParams,
        GetOrderHistoryParamsBuilder, GetOrderListParams, GetOrderListParamsBuilder,
        GetOrderParams, GetOrderParamsBuilder, GetPositionTiersParams, GetPositionsHistoryParams,
        GetPositionsParams, GetPositionsParamsBuilder, GetPriceLimitParams,
        GetPriceLimitParamsBuilder, GetRpiOrderBookParams, GetSpreadOrderParams,
        GetSpreadOrderParamsBuilder, GetSpreadOrdersParams, GetSpreadOrdersParamsBuilder,
        GetSpreadTradesParams, GetSpreadTradesParamsBuilder, GetSpreadsParams, 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_POST_ONLY_CANCEL_REASON, OKX_POST_ONLY_CANCEL_SOURCE, OKX_SUPPORTED_ORDER_TYPES,
            OKX_SUPPORTED_TIME_IN_FORCE, okx_algo_target_currency, okx_reduce_only_wire_value,
            spot_trade_quote_ccy_wire_value, validate_algo_trigger_type,
        },
        credential::Credential,
        enums::{
            OKXAlgoOrderStatus, OKXAlgoOrderType, OKXEnvironment, OKXInstrumentStatus,
            OKXInstrumentType, OKXMarginMode, OKXOrderStatus, OKXOrderType, OKXPositionMode,
            OKXPositionSide, OKXSide, OKXTargetCurrency, OKXTradeMode, OKXTriggerType,
            conditional_order_to_algo_type,
        },
        models::OKXInstrument,
        parse::{
            extract_inst_family, is_okx_spread_symbol, is_order_status_report_more_advanced,
            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_mark_price_update,
            parse_millisecond_timestamp, parse_order_status_report, parse_position_status_report,
            parse_price, parse_quantity, parse_spot_margin_position_from_balance,
            parse_spread_fill_report, parse_spread_instrument, parse_spread_order_status_report,
            parse_trade_tick, prefer_rpi_response_fields,
        },
    },
    http::models::{OKXCandlestick, OKXTrade},
    websocket::{messages::OKXAlgoOrderMsg, parse::parse_algo_order_status_report},
};

const OKX_SUCCESS_CODE: &str = "0";
const OKX_PARTIAL_SUCCESS_CODE: &str = "2";
const RETRY_AFTER_HEADER: &str = "Retry-After";

#[derive(Debug, Error)]
#[error("Failed to parse instrument {symbol}: {source}")]
pub(crate) struct OKXInstrumentDefinitionError {
    symbol: String,
    #[source]
    source: anyhow::Error,
}

impl OKXInstrumentDefinitionError {
    fn new(symbol: &str, source: anyhow::Error) -> Self {
        Self {
            symbol: symbol.to_string(),
            source,
        }
    }
}

#[derive(Debug, Error)]
#[error("Failed to fetch pending algo order reports: {source}")]
pub(crate) struct OKXPendingAlgoOrderReportsError {
    #[source]
    source: anyhow::Error,
}

impl OKXPendingAlgoOrderReportsError {
    fn new(source: anyhow::Error) -> Self {
        Self { source }
    }
}

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

// OKX omits `data` from gateway errors such as authentication failures
#[derive(Deserialize)]
struct OKXErrorEnvelope {
    code: String,
    msg: String,
}

fn resolve_okx_error_code(response_body: &[u8], envelope_code: &str) -> String {
    if let Ok(payload) = serde_json::from_slice::<serde_json::Value>(response_body)
        && let Some(s_code) = payload
            .get("data")
            .and_then(serde_json::Value::as_array)
            .and_then(|items| items.first())
            .and_then(|item| 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();
        }
    }

    envelope_code.to_string()
}

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()
}

fn deserialize_okx_response<T: DeserializeOwned>(
    response_body: &[u8],
) -> Result<OKXResponse<T>, OKXHttpError> {
    if response_body.is_empty() {
        return Err(OKXHttpError::EmptyResponse);
    }

    let contains_legacy_rpi_name = [br#""elp""#.as_slice(), br#""elpMaker""#.as_slice()]
        .iter()
        .any(|name| {
            response_body
                .windows(name.len())
                .any(|window| window == *name)
        });

    if !contains_legacy_rpi_name {
        return serde_json::from_slice(response_body)
            .map_err(|e| classify_response_decode_error(&e));
    }

    let mut value: serde_json::Value =
        serde_json::from_slice(response_body).map_err(|e| classify_response_decode_error(&e))?;
    prefer_rpi_response_fields(&mut value);
    serde_json::from_value(value).map_err(|e| classify_response_decode_error(&e))
}

fn classify_response_decode_error(error: &serde_json::Error) -> OKXHttpError {
    if error.is_data() {
        OKXHttpError::ResponseDecoding(error.to_string())
    } else {
        OKXHttpError::MalformedResponse(error.to_string())
    }
}

fn retry_after(headers: &HashMap<String, String>, now: Timestamp) -> Option<Duration> {
    let value = headers.get(RETRY_AFTER_HEADER)?;
    let delay = parse_retry_after(value, now);
    if delay.is_none() {
        log::warn!("Invalid OKX response header {RETRY_AFTER_HEADER}={value:?}");
    }
    delay
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use anyhow::Context;
    use nautilus_model::{
        identifiers::{InstrumentId, Symbol},
        instruments::{
            CryptoFuture, CryptoFuturesSpread, InstrumentAny,
            stubs::{crypto_future_btcusdt, crypto_futures_spread_btc_deribit},
        },
    };
    use nautilus_testkit::http::assert_http_redirect_rejected;
    use rstest::rstest;
    use rust_decimal::Decimal;
    use serde::{Serialize, Serializer, ser::Error as _};

    use super::{
        Method, OKX_GLOBAL_RATE_KEY, OKXEnvironment, OKXHttpClient, OKXHttpError,
        OKXInstrumentDefinitionError, OKXRawHttpClient, Timestamp, deserialize_okx_response,
        parse_retry_after, resolve_okx_error_code, resolve_okx_error_message,
    };
    use crate::http::models::OKXFeeRate;

    #[rstest]
    fn test_is_spread_prefers_cached_future(mut crypto_future_btcusdt: CryptoFuture) {
        let client = OKXHttpClient::default();
        let instrument_id = InstrumentId::from("BTC-USDT_BTC-USDT-SWAP.OKX");
        crypto_future_btcusdt.id = instrument_id;
        crypto_future_btcusdt.raw_symbol = instrument_id.symbol;
        let uncached = client.is_spread(instrument_id);

        client.cache_instrument(InstrumentAny::CryptoFuture(crypto_future_btcusdt));

        assert!(uncached);
        assert!(!client.is_spread(instrument_id));
    }

    #[rstest]
    fn test_is_spread_prefers_cached_spread(
        mut crypto_futures_spread_btc_deribit: CryptoFuturesSpread,
    ) {
        let client = OKXHttpClient::default();
        let instrument_id = InstrumentId::from("BTC-USD_UM_XPERP-310404.OKX");
        crypto_futures_spread_btc_deribit.id = instrument_id;
        crypto_futures_spread_btc_deribit.raw_symbol = Symbol::from("BTC-USD_UM_XPERP-310404");
        let uncached = client.is_spread(instrument_id);

        client.cache_instrument(InstrumentAny::CryptoFuturesSpread(
            crypto_futures_spread_btc_deribit,
        ));

        assert!(!uncached);
        assert!(client.is_spread(instrument_id));
    }

    struct UnserializableParams;

    impl Serialize for UnserializableParams {
        fn serialize<S>(&self, _serializer: S) -> Result<S::Ok, S::Error>
        where
            S: Serializer,
        {
            Err(S::Error::custom("intentional serialization failure"))
        }
    }

    #[tokio::test]
    async fn test_authenticated_client_rejects_redirects() {
        let client = OKXRawHttpClient::with_credentials(
            "key".into(),
            "secret".into(),
            "pass".into(),
            "http://localhost".into(),
            3,
            0,
            1,
            1,
            OKXEnvironment::Demo,
            None,
        )
        .unwrap()
        .client;
        assert_http_redirect_rejected(|url| async move {
            client
                .get(url, None, None, Some(3), None)
                .await
                .unwrap()
                .status
                .as_u16()
        })
        .await;
    }

    #[rstest]
    fn test_instrument_definition_error_survives_anyhow_context_downcast() {
        let result: anyhow::Result<()> = Err(OKXInstrumentDefinitionError::new(
            "USDG-SGD",
            anyhow::anyhow!("`tick_sz` is empty"),
        )
        .into());

        let err = result.context("fetch instrument from API").unwrap_err();

        assert!(err.downcast_ref::<OKXInstrumentDefinitionError>().is_some());
    }

    #[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]
    fn test_resolve_okx_error_code_prefers_item_s_code_over_envelope() {
        let body = br#"{
            "code": "1",
            "msg": "All operations failed",
            "data": [
                {
                    "sCode": "50013",
                    "sMsg": "System busy, please retry later"
                }
            ]
        }"#;

        assert_eq!(resolve_okx_error_code(body, "1"), "50013");
    }

    #[rstest]
    #[case("/api/v5/account/bills", 200)]
    #[case("/api/v5/account/bills-archive", 400)]
    fn test_account_bills_quotas(#[case] endpoint: &str, #[case] interval_ms: u64) {
        let keys = OKXRawHttpClient::rate_limit_keys(&format!("{endpoint}?ccy=USDT"));
        assert_eq!(keys[0].as_str(), OKX_GLOBAL_RATE_KEY);
        assert_eq!(keys[1].as_str(), format!("okx:{endpoint}"));
        let quota = OKXRawHttpClient::rate_limiter_quotas()
            .into_iter()
            .find_map(|(key, quota)| (key == keys[1].as_str()).then_some(quota))
            .expect("account bills quota");

        // Pace requests without a burst to stay within each endpoint's five-request window
        assert_eq!(quota.burst_size().get(), 1);
        assert_eq!(
            quota.replenish_interval(),
            Duration::from_millis(interval_ms)
        );
    }

    #[rstest]
    fn test_activate_feature_quota_fits_five_per_two_seconds() {
        let keys = OKXRawHttpClient::rate_limit_keys("/api/v5/account/activate-feature");
        let quota = OKXRawHttpClient::rate_limiter_quotas()
            .into_iter()
            .find_map(|(key, quota)| (key == keys[1].as_str()).then_some(quota))
            .expect("activate-feature quota");

        assert_eq!(quota.burst_size().get(), 2);
        assert_eq!(quota.replenish_interval(), Duration::from_millis(500));
    }

    #[rstest]
    fn test_rpi_response_fields_prefer_current_names_over_transition_aliases() {
        let response = br#"{
            "code": "0",
            "msg": "",
            "data": [{
                "level": "VIP1",
                "taker": "-0.0005",
                "maker": "-0.0002",
                "takerU": "-0.0005",
                "makerU": "-0.0002",
                "rpiMaker": "-0.00015",
                "elpMaker": "-0.00016",
                "instType": "SPOT",
                "category": "1",
                "ts": "1785406500000"
            }]
        }"#;

        let parsed = deserialize_okx_response::<OKXFeeRate>(response).unwrap();

        assert_eq!(parsed.data.len(), 1);
        assert_eq!(parsed.data[0].rpi_maker, Some(Decimal::new(-15, 5)));
    }

    #[rstest]
    fn test_parse_retry_after_supports_delay_seconds_and_http_date() {
        let now: Timestamp = "1994-11-06T08:49:36Z".parse().unwrap();

        assert_eq!(
            parse_retry_after("5", now),
            Some(std::time::Duration::from_secs(5))
        );
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 08:51:36 GMT", now),
            Some(std::time::Duration::from_secs(120))
        );
        assert_eq!(
            parse_retry_after("Sun, 06 Nov 1994 08:47:36 GMT", now),
            Some(std::time::Duration::ZERO)
        );
        assert_eq!(parse_retry_after("soon", now), None);
    }

    #[rstest]
    fn test_empty_response_is_distinct_from_malformed_json() {
        let empty = deserialize_okx_response::<serde_json::Value>(b"").unwrap_err();
        let malformed = deserialize_okx_response::<serde_json::Value>(b"{").unwrap_err();

        assert!(matches!(empty, OKXHttpError::EmptyResponse));
        assert!(matches!(malformed, OKXHttpError::MalformedResponse(_)));
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_serialization_failure_is_typed_before_transport() {
        let client = OKXRawHttpClient::new(
            Some("http://127.0.0.1:1".to_string()),
            1,
            0,
            1,
            1,
            OKXEnvironment::Live,
            None,
        )
        .unwrap();

        let error = client
            .send_request::<serde_json::Value, _>(
                Method::GET,
                "/api/v5/test",
                Some(&UnserializableParams),
                None,
                false,
            )
            .await
            .unwrap_err();

        assert!(matches!(
            error,
            OKXHttpError::RequestSerialization(message)
                if message.contains("intentional serialization failure")
        ));
    }

    #[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;

#[derive(Clone, Copy, Debug)]
pub(crate) struct ReportInstrumentScope<'a> {
    pub instrument_types: &'a [OKXInstrumentType],
    pub load_spreads: bool,
    pub margin_mode: Option<OKXMarginMode>,
}

impl ReportInstrumentScope<'_> {
    pub(crate) fn validate_position_margin(&self, position: &OKXPosition) -> anyhow::Result<()> {
        if let Some(expected) = self.margin_mode
            && matches!(
                position.inst_type,
                OKXInstrumentType::Swap | OKXInstrumentType::Futures
            )
            && matches!(
                position.pos_side,
                OKXPositionSide::Long | OKXPositionSide::Short
            )
            && position.mgn_mode != expected
            && !Decimal::from_str(&position.pos)?.is_zero()
        {
            anyhow::bail!(
                "OKX long/short position {} uses margin mode {}, expected {expected}; mixed margin modes are not supported",
                position.inst_id,
                position.mgn_mode,
            );
        }

        Ok(())
    }
}

#[derive(Debug)]
pub(crate) struct ReportSweep<T> {
    pub reports: Vec<T>,
    pub complete: bool,
}

#[derive(Clone, Copy)]
pub(crate) enum FillHistory {
    Recent,
    Extended,
}

#[derive(Debug)]
pub(crate) struct AlgoOrderReportSweep {
    pub reports: Vec<OrderStatusReport>,
    pub complete: bool,
    pub ambiguous_triggered_child_ids: AHashSet<VenueOrderId>,
}

struct PageSweep<T> {
    items: Vec<T>,
    complete: bool,
}

impl<T> PageSweep<T> {
    fn from_pages(items: Vec<T>, exhausted: bool) -> Self {
        Self {
            complete: !exhausted,
            items,
        }
    }
}

enum InstrumentResolution {
    Found(Box<InstrumentAny>),
    Skip,
    Incomplete,
}

/// 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 {
    clock: &'static AtomicTime,
    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/config".to_string(),
                Quota::per_second(NonZeroU32::new(2).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/set-position-mode".to_string(),
                Quota::per_second(NonZeroU32::new(2).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/activate-feature".to_string(),
                Quota::per_second(NonZeroU32::new(2).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/balance".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/bills".to_string(),
                Quota::with_period(Duration::from_millis(200)).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/bills-archive".to_string(),
                Quota::with_period(Duration::from_millis(400)).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/trade-fee".to_string(),
                Quota::per_second(NonZeroU32::new(2).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/instruments".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/positions".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/account/positions-history".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/public/position-tiers".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/event-contract/series".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/event-contract/events".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/event-contract/markets".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/opt-summary".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/time".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/mark-price".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/price-limit".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/spreads".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/order".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/cancel-order".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/mass-cancel".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/orders-pending".to_string(),
                Quota::per_second(NonZeroU32::new(5).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/orders-history".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/sprd/trades".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/index-tickers".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/books".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/books-rpi".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/candles".to_string(),
                Quota::per_second(NonZeroU32::new(20).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/history-candles".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/market/history-trades".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/public/funding-rate-history".to_string(),
                Quota::per_second(NonZeroU32::new(5).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"),
            ),
            (
                "okx:/api/v5/trade/batch-orders".to_string(),
                Quota::per_second(NonZeroU32::new(7).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/amend-order".to_string(),
                Quota::per_second(NonZeroU32::new(30).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/amend-batch-orders".to_string(),
                Quota::per_second(NonZeroU32::new(7).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/cancel-batch-orders".to_string(),
                Quota::per_second(NonZeroU32::new(7).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/orders-pending".to_string(),
                Quota::per_second(NonZeroU32::new(30).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/fills-history".to_string(),
                Quota::per_second(NonZeroU32::new(5).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(1).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/cancel-advance-algos".to_string(),
                Quota::per_second(NonZeroU32::new(1).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"),
            ),
            (
                "okx:/api/v5/trade/orders-algo-pending".to_string(),
                Quota::per_second(NonZeroU32::new(10).expect("non-zero")).expect("valid constant"),
            ),
            (
                "okx:/api/v5/trade/orders-algo-history".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 {
            clock: get_atomic_clock_realtime(),
            base_url: base_url.unwrap_or(OKX_HTTP_URL.to_string()),
            client: HttpClient::builder()
                .headers(Self::default_headers(environment))
                .header_keys(vec![RETRY_AFTER_HEADER.to_string()])
                .keyed_quotas(Self::rate_limiter_quotas())
                .default_quota(*OKX_REST_QUOTA)
                .timeout_secs(timeout_secs)
                .maybe_proxy_url(proxy_url)
                .build()
                .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 {
            clock: get_atomic_clock_realtime(),
            base_url,
            client: HttpClient::builder()
                .redirect_policy(HttpRedirectPolicy::Reject)
                .headers(Self::default_headers(environment))
                .header_keys(vec![RETRY_AFTER_HEADER.to_string()])
                .keyed_quotas(Self::rate_limiter_quotas())
                .default_quota(*OKX_REST_QUOTA)
                .timeout_secs(timeout_secs)
                .maybe_proxy_url(proxy_url)
                .build()
                .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<String, String> =
            create_standard_nautilus_headers().into_iter().collect();

        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 an error if no credentials are set but the request requires authentication.
    fn sign_request(
        &self,
        method: &Method,
        path: &str,
        body: Option<&[u8]>,
        now: Timestamp,
    ) -> 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 timestamp = format!("{now:.3}");
        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 query_string = params
            .map(serde_urlencoded::to_string)
            .transpose()
            .map_err(|e| {
                OKXHttpError::RequestSerialization(format!("Failed to serialize params: {e}"))
            })?
            .unwrap_or_default();
        let full_path = if query_string.is_empty() {
            path.to_string()
        } else {
            format!("{path}?{query_string}")
        };
        let url = format!("{}{full_path}", self.base_url);
        let accepts_partial_success = matches!(
            path,
            "/api/v5/trade/batch-orders"
                | "/api/v5/trade/amend-batch-orders"
                | "/api/v5/trade/cancel-batch-orders"
        );

        // 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();
            let full_path = full_path.clone();

            async move {
                let mut headers = if authenticate {
                    let now = self.clock.get_time_ns().to_datetime_utc();
                    self.sign_request(&method, &full_path, body.as_deref(), now)?
                } 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,
                        None,
                        Some(headers),
                        body,
                        None,
                        Some(rate_keys),
                    )
                    .await?;

                log::trace!("Response: {resp:?}");
                let now = self.clock.get_time_ns().to_datetime_utc();
                let retry_after = retry_after(&resp.headers, now);

                if resp.status.is_success() {
                    let okx_response: OKXResponse<T> = deserialize_okx_response(&resp.body)?;

                    if okx_response.code != OKX_SUCCESS_CODE
                        && !(accepts_partial_success
                            && okx_response.code == OKX_PARTIAL_SUCCESS_CODE)
                    {
                        let error_code = resolve_okx_error_code(&resp.body, &okx_response.code);
                        let message = resolve_okx_error_message(&resp.body, &okx_response.msg);
                        return Err(OKXHttpError::from_venue_response(
                            error_code,
                            message,
                            retry_after,
                        ));
                    }

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

                    if let Ok(envelope) = serde_json::from_slice::<OKXErrorEnvelope>(&resp.body) {
                        let error_code = resolve_okx_error_code(&resp.body, &envelope.code);
                        let message = resolve_okx_error_message(&resp.body, &envelope.msg);
                        return Err(OKXHttpError::from_venue_response(
                            error_code,
                            message,
                            retry_after,
                        ));
                    }

                    if status.as_u16() >= 500 || status == StatusCode::TOO_MANY_REQUESTS {
                        Err(OKXHttpError::RetryableStatus {
                            status,
                            body: error_body.to_string(),
                            retry_after,
                        })
                    } else {
                        Err(OKXHttpError::UnexpectedStatus {
                            status,
                            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. Undecoded 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")
        //
        // Submit POSTs are exempt: OKX rejects a duplicate `clOrdId` only while
        // the first order rests open, so an attempt whose response was lost can
        // already have filled and freed the client order ID. Retrying could then
        // place a second live order, so submits are sent once and an ambiguous
        // outcome is left for stream updates and reconciliation to resolve.
        // Algo amendments also run once: a retry can reject after an earlier
        // attempt applied and the order triggered, leaving the original outcome ambiguous.
        let is_order_mutation = method == Method::POST
            && matches!(
                path,
                "/api/v5/trade/order"
                    | "/api/v5/trade/batch-orders"
                    | "/api/v5/trade/order-algo"
                    | "/api/v5/trade/amend-algos"
                    | "/api/v5/sprd/order"
            );
        let should_retry = |error: &OKXHttpError| !is_order_mutation && error.is_retryable();

        let create_error = |error: RetryError| -> OKXHttpError {
            match error {
                RetryError::Canceled => {
                    OKXHttpError::Canceled("Adapter disconnecting or shutting down".to_string())
                }
                RetryError::OperationTimeout { timeout_ms } => {
                    OKXHttpError::OperationTimeout { timeout_ms }
                }
                RetryError::InvalidConfiguration { message } => {
                    OKXHttpError::ValidationError(message)
                }
                error @ RetryError::ElapsedBudgetExceeded { .. } => {
                    OKXHttpError::RetryBudgetExceeded(error.to_string())
                }
            }
        };

        self.retry_manager
            .invocation(path, operation, should_retry, create_error)
            .retry_delay(&OKXHttpError::retry_after)
            .cancellation_token(&self.cancellation_token)
            .execute()
            .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
    }

    /// Activates an account feature such as USDC order book trading.
    ///
    /// This method does not run implicitly. Call it only after order placement
    /// returns error code `54109`. Activation is shared between a master account and
    /// its sub-accounts, so one successful call covers all of them. Error code
    /// `51773` means the account does not support activation, not that USDC trading
    /// is unavailable.
    ///
    /// # 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/log_en/#upcoming-changes-okx-to-migrate-usd-spot-trading-pairs-new-endpoint-activate-usdc-trading>
    pub async fn activate_feature(
        &self,
        params: ActivateFeatureParams,
    ) -> Result<Vec<serde_json::Value>, OKXHttpError> {
        let path = "/api/v5/account/activate-feature";
        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 account instrument configuration and trading permissions.
    ///
    /// # Errors
    ///
    /// Returns an error if authentication fails, the HTTP request fails, or the
    /// response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-instruments>
    pub async fn get_account_instruments(
        &self,
        params: GetInstrumentsParams,
    ) -> Result<Vec<OKXInstrument>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/instruments",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests a list of spread trading instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or if the response body cannot
    /// be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-get-spreads-public>
    pub async fn get_spreads(
        &self,
        params: GetSpreadsParams,
    ) -> Result<Vec<OKXSpread>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/sprd/spreads",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Places a spread order.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-post-place-order>
    pub async fn place_spread_order(
        &self,
        request: OKXPlaceSpreadOrderRequest,
    ) -> Result<Vec<OKXPlaceOrderResponse>, OKXHttpError> {
        let body = serde_json::to_vec(&request)?;

        self.send_request(
            Method::POST,
            "/api/v5/sprd/order",
            None::<&()>,
            Some(body),
            true,
        )
        .await
    }

    /// Cancels a spread order.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-post-cancel-order>
    pub async fn cancel_spread_order(
        &self,
        request: OKXCancelSpreadOrderRequest,
    ) -> Result<Vec<OKXCancelOrderResponse>, OKXHttpError> {
        let body = serde_json::to_vec(&request)?;

        self.send_request(
            Method::POST,
            "/api/v5/sprd/cancel-order",
            None::<&()>,
            Some(body),
            true,
        )
        .await
    }

    /// Cancels all orders for a spread.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-post-mass-cancel>
    pub async fn cancel_all_spread_orders(
        &self,
        request: OKXCancelAllSpreadOrdersRequest,
    ) -> Result<Vec<OKXCancelOrderResponse>, OKXHttpError> {
        let body = serde_json::to_vec(&request)?;

        self.send_request(
            Method::POST,
            "/api/v5/sprd/mass-cancel",
            None::<&()>,
            Some(body),
            true,
        )
        .await
    }

    /// Requests spread order details.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-get-order-details>
    pub async fn get_spread_order(
        &self,
        params: GetSpreadOrderParams,
    ) -> Result<Vec<OKXSpreadOrder>, OKXHttpError> {
        self.send_request(Method::GET, "/api/v5/sprd/order", Some(&params), None, true)
            .await
    }

    /// Requests pending spread orders.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-get-active-orders>
    pub async fn get_spread_orders_pending(
        &self,
        params: GetSpreadOrdersParams,
    ) -> Result<Vec<OKXSpreadOrder>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/sprd/orders-pending",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests historical spread orders.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-get-orders-history-last-3-months>
    pub async fn get_spread_orders_history(
        &self,
        params: GetSpreadOrdersParams,
    ) -> Result<Vec<OKXSpreadOrder>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/sprd/orders-history",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests spread trades.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response cannot be deserialized.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#spread-trading-rest-api-get-trades-last-7-days>
    pub async fn get_spread_trades(
        &self,
        params: GetSpreadTradesParams,
    ) -> Result<Vec<OKXSpreadTrade>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/sprd/trades",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests OKX event contract series.
    ///
    /// # 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-series>.
    pub async fn get_event_contract_series(
        &self,
        params: GetEventContractSeriesParams,
    ) -> Result<Vec<OKXEventContractSeries>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/event-contract/series",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests OKX event contract events for a series.
    ///
    /// # 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-events>.
    pub async fn get_event_contract_events(
        &self,
        params: GetEventContractEventsParams,
    ) -> Result<Vec<OKXEventContractEvent>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/event-contract/events",
            Some(&params),
            None,
            false,
        )
        .await
    }

    /// Requests OKX event contract markets for a series.
    ///
    /// # 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-markets>.
    pub async fn get_event_contract_markets(
        &self,
        params: GetEventContractMarketsParams,
    ) -> Result<Vec<OKXEventContractMarket>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/event-contract/markets",
            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(OKXHttpError::EmptyResponse)
    }

    /// 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 current price limits for an instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or if the response body
    /// cannot be parsed into [`OKXPriceLimit`].
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#public-data-rest-api-get-limit-price>
    pub async fn get_price_limit(
        &self,
        params: GetPriceLimitParams,
    ) -> Result<Vec<OKXPriceLimit>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/public/price-limit",
            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 a Retail Price Improvement 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-rpi-order-book>
    pub async fn get_rpi_order_book(
        &self,
        params: GetRpiOrderBookParams,
    ) -> Result<Vec<OKXRpiOrderBookSnapshot>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/market/books-rpi",
            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 the authenticated account's configuration.
    ///
    /// Returns the response data array, which is empty if OKX returns no configuration.
    /// This query exposes exchange configuration without applying account or permission policy.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, OKX returns an error,
    /// or required configuration fields are missing or invalid.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-account-configuration>
    pub async fn get_account_configuration(
        &self,
    ) -> Result<Vec<OKXAccountConfiguration>, OKXHttpError> {
        let path = "/api/v5/account/config";
        self.send_request::<_, ()>(Method::GET, path, None, None, true)
            .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 one page of trading-account bills from the last seven days.
    ///
    /// Returns bills in venue order (most recent first), including non-trade balance changes.
    /// Pagination and historical coverage remain the caller's responsibility.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, OKX returns an error,
    /// or the response cannot be decoded.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-bills-details-last-7-days>
    pub async fn get_account_bills(
        &self,
        params: GetAccountBillsParams,
    ) -> Result<Vec<OKXAccountBill>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/bills",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests one page of trading-account bills from the last three months.
    ///
    /// Returns bills in venue order (most recent first), including non-trade balance changes.
    /// Pagination and historical coverage remain the caller's responsibility.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the request fails, OKX returns an error,
    /// or the response cannot be decoded.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#trading-account-rest-api-get-bills-details-last-3-months>
    pub async fn get_account_bills_archive(
        &self,
        params: GetAccountBillsParams,
    ) -> Result<Vec<OKXAccountBill>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/account/bills-archive",
            Some(&params),
            None,
            true,
        )
        .await
    }

    /// Requests fee rates for the account.
    ///
    /// Returns fee rates for the specified instrument type and the user's VIP level.
    /// Incentive-program rates require `inst_id` for SPOT/MARGIN or `inst_family` for
    /// FUTURES/SWAP/OPTION; a `group_id` query alone does not provide equivalent rates.
    /// Fee selection remains with the caller. The endpoint does not reflect zero-fee promotions.
    ///
    /// # 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> {
        if params.group_id.is_some() && (params.inst_id.is_some() || params.inst_family.is_some()) {
            return Err(OKXHttpError::ValidationError(
                "group_id cannot be combined with inst_id or inst_family".to_string(),
            ));
        }

        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
    }

    /// Retrieves a single algo order's details.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/en/#order-book-trading-algo-trading-get-algo-order-details>
    pub async fn get_algo_order(
        &self,
        params: GetAlgoOrderParams,
    ) -> Result<Vec<OKXOrderAlgo>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/order-algo",
            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 transaction details (fills) from the extended history.
    ///
    /// # 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-months>
    pub async fn get_fills_history(
        &self,
        params: GetTransactionDetailsParams,
    ) -> Result<Vec<OKXTransactionDetail>, OKXHttpError> {
        self.send_request(
            Method::GET,
            "/api/v5/trade/fills-history",
            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.adapters.okx", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.okx")
)]
pub struct OKXHttpClient {
    pub(crate) inner: Arc<OKXRawHttpClient>,
    pub(crate) instruments_cache: Arc<AtomicMap<Ustr, InstrumentAny>>,
    trade_quote_ccy_lists: Arc<AtomicMap<Ustr, Vec<Ustr>>>,
    spot_trade_quote_ccy: Arc<Mutex<Option<Ustr>>>,
    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(),
            trade_quote_ccy_lists: self.trade_quote_ccy_lists.clone(),
            spot_trade_quote_ccy: self.spot_trade_quote_ccy.clone(),
            cache_initialized,
        }
    }
}

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()),
            trade_quote_ccy_lists: Arc::new(AtomicMap::new()),
            spot_trade_quote_ccy: Arc::new(Mutex::new(None)),
            cache_initialized: AtomicBool::new(false),
        })
    }

    /// Generates a timestamp for initialization.
    fn generate_ts_init(&self) -> UnixNanos {
        self.inner.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()),
            trade_quote_ccy_lists: Arc::new(AtomicMap::new()),
            spot_trade_quote_ccy: Arc::new(Mutex::new(None)),
            cache_initialized: AtomicBool::new(false),
        })
    }

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

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

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

    /// Requests order list (pending orders).
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub async fn get_orders_pending(
        &self,
        params: GetOrderListParams,
    ) -> Result<Vec<OKXOrderHistory>, OKXHttpError> {
        self.inner.get_orders_pending(params).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.inner.get_order_algo_pending(params).await
    }

    /// Requests information on current account positions.
    ///
    /// # Errors
    ///
    /// Returns an error if the operation fails.
    pub async fn get_positions(
        &self,
        params: GetPositionsParams,
    ) -> Result<Vec<OKXPosition>, OKXHttpError> {
        self.inner.get_positions(params).await
    }

    /// 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(Credential::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(Credential::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(ToString::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)
    }

    pub(crate) fn is_spread(&self, instrument_id: InstrumentId) -> bool {
        self.instruments_cache
            .load()
            .get(&instrument_id.symbol.inner())
            .map_or_else(
                || is_okx_spread_symbol(instrument_id.symbol.as_str()),
                InstrumentAny::is_spread,
            )
    }

    /// Sets the optional SPOT `tradeQuoteCcy` override for subsequent order placement.
    pub fn set_spot_trade_quote_ccy(&self, ccy: Option<String>) {
        *self
            .spot_trade_quote_ccy
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) =
            ccy.map(|value| Ustr::from(value.as_str()));
    }

    /// Caches `tradeQuoteCcyList` values keyed by instrument ID.
    pub fn cache_trade_quote_ccy_lists(
        &self,
        mappings: impl IntoIterator<Item = (Ustr, Vec<Ustr>)>,
    ) {
        let entries: Vec<_> = mappings.into_iter().collect();
        self.trade_quote_ccy_lists.rcu(|m| {
            for (inst_id, list) in &entries {
                m.insert(*inst_id, list.clone());
            }
        });
    }

    /// Returns a snapshot of cached `tradeQuoteCcyList` values.
    #[must_use]
    pub fn trade_quote_ccy_lists_snapshot(&self) -> Vec<(Ustr, Vec<Ustr>)> {
        self.trade_quote_ccy_lists
            .load()
            .iter()
            .map(|(inst_id, list)| (*inst_id, list.clone()))
            .collect()
    }

    fn cache_trade_quote_ccy_list_from_instrument(&self, instrument: &OKXInstrument) {
        if !instrument.trade_quote_ccy_list.is_empty() {
            self.trade_quote_ccy_lists
                .insert(instrument.inst_id, instrument.trade_quote_ccy_list.clone());
        }
    }

    fn resolve_spot_trade_quote_ccy(
        &self,
        instrument_type: OKXInstrumentType,
        inst_id: Ustr,
    ) -> Result<Option<Ustr>, OKXHttpError> {
        let configured = *self
            .spot_trade_quote_ccy
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let available = self
            .trade_quote_ccy_lists
            .load()
            .get(&inst_id)
            .cloned()
            .unwrap_or_default();
        spot_trade_quote_ccy_wire_value(
            instrument_type,
            configured.as_ref().map(Ustr::as_str),
            &available,
        )
        .map_err(OKXHttpError::ValidationError)
    }

    /// Requests the account state for the `account_id` from OKX.
    ///
    /// Pass the execution client's configured account type; the OKX balance payload carries
    /// no account-mode field.
    ///
    /// # 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,
        account_type: AccountType,
    ) -> 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, account_type, 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)
            }
        }
    }

    /// Activates an account feature such as USDC order book trading.
    ///
    /// This does not run at client start. Call it only after OKX rejects a
    /// `Crypto-USDC` order with error code `54109`. Activation is shared between a
    /// master account and its sub-accounts, so one successful call covers all of
    /// them. Error code `51773` means the account does not support activation, not
    /// that USDC trading is unavailable; a successful order confirms the account can
    /// trade the instrument.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails.
    ///
    /// # References
    ///
    /// <https://www.okx.com/docs-v5/log_en/#upcoming-changes-okx-to-migrate-usd-spot-trading-pairs-new-endpoint-activate-usdc-trading>
    pub async fn activate_feature(&self, feature: &str) -> anyhow::Result<()> {
        let params = ActivateFeatureParams {
            feature: feature.to_string(),
        };

        self.inner
            .activate_feature(params)
            .await
            .map(|_| ())
            .map_err(Into::into)
    }

    /// Refreshes cached `tradeQuoteCcyList` values from private instruments.
    ///
    /// # Errors
    ///
    /// Returns an error if the authenticated instruments request fails.
    pub async fn refresh_account_trade_quote_ccy_lists(
        &self,
        instrument_type: OKXInstrumentType,
        instrument_family: Option<String>,
    ) -> anyhow::Result<()> {
        let mut params = GetInstrumentsParamsBuilder::default();
        params.inst_type(instrument_type);
        if let Some(family) = instrument_family {
            params.inst_family(family);
        }

        let params = params.build().map_err(|e| anyhow::anyhow!(e))?;
        let instruments = self
            .inner
            .get_account_instruments(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;
        self.cache_trade_quote_ccy_lists(
            instruments
                .iter()
                .filter(|instrument| !instrument.trade_quote_ccy_list.is_empty())
                .map(|instrument| (instrument.inst_id, instrument.trade_quote_ccy_list.clone())),
        );

        Ok(())
    }

    /// Requests all instruments for the `instrument_type` from OKX.
    ///
    /// Option requests require `instrument_family` (OKX `instFamily`), for example `BTC-USD`.
    ///
    /// # Errors
    ///
    /// Returns an error if `instrument_type` is option and `instrument_family` is missing,
    /// 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)>)> {
        if instrument_type == OKXInstrumentType::Option && instrument_family.is_none() {
            anyhow::bail!(
                "option instruments require instrument_family (OKX instFamily), for example BTC-USD"
            );
        }

        let resp = if instrument_type == OKXInstrumentType::Events {
            let series_ids = if let Some(series_id) = instrument_family.clone() {
                vec![series_id]
            } else {
                self.inner
                    .get_event_contract_series(GetEventContractSeriesParams::default())
                    .await
                    .map_err(|e| anyhow::anyhow!(e))?
                    .into_iter()
                    .map(|series| series.series_id)
                    .collect()
            };

            let mut event_instruments = Vec::new();

            for series_id in series_ids {
                let mut params = GetInstrumentsParamsBuilder::default();
                params.inst_type(OKXInstrumentType::Events);
                params.series_id(series_id);
                let params = params.build().map_err(|e| anyhow::anyhow!(e))?;
                let mut page = self
                    .inner
                    .get_instruments(params)
                    .await
                    .map_err(|e| anyhow::anyhow!(e))?;
                event_instruments.append(&mut page);
            }
            event_instruments
        } else {
            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))?;

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

        let ts_init = self.generate_ts_init();

        let mut instruments: Vec<InstrumentAny> = Vec::new();
        let mut inst_id_codes: Vec<(Ustr, u64)> = Vec::new();
        let mut trade_quote_ccy_lists: Vec<(Ustr, Vec<Ustr>)> = 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));
            }

            if !inst.trade_quote_ccy_list.is_empty() {
                trade_quote_ccy_lists.push((inst.inst_id, inst.trade_quote_ccy_list.clone()));
            }

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

            match parse_instrument_any(inst, None, None, 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);
                }
            }
        }

        self.cache_trade_quote_ccy_lists(trade_quote_ccy_lists);

        Ok((instruments, inst_id_codes))
    }

    /// Requests spread instruments from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or spread parsing fails.
    pub async fn request_spread_instruments(
        &self,
        params: GetSpreadsParams,
    ) -> anyhow::Result<Vec<InstrumentAny>> {
        let resp = self
            .inner
            .get_spreads(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

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

        for spread in &resp {
            match parse_spread_instrument(spread, None, None, ts_init) {
                Ok(instrument) => instruments.push(instrument),
                Err(e) => log::warn!("Failed to parse spread {}: {e}", spread.sprd_id),
            }
        }

        Ok(instruments)
    }

    /// Requests a single instrument by `instrument_id` from OKX.
    ///
    /// Fetches the instrument from the API, caches it, and returns it.
    ///
    /// # Errors
    ///
    /// Returns 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();
        if self.is_spread(instrument_id) {
            let instrument = self.request_spread_instrument(symbol).await?;
            self.cache_instrument(instrument.clone());
            return Ok(instrument);
        }

        let instrument_type = okx_instrument_type_from_symbol(symbol);

        let resp = if instrument_type == OKXInstrumentType::Events {
            let series = self
                .inner
                .get_event_contract_series(GetEventContractSeriesParams::default())
                .await
                .map_err(|e| anyhow::anyhow!(e))?;

            let mut matched = Vec::new();

            for series in series {
                let mut params = GetInstrumentsParamsBuilder::default();
                params.inst_type(OKXInstrumentType::Events);
                params.series_id(series.series_id);
                params.inst_id(symbol);
                let params = params.build().map_err(|e| anyhow::anyhow!(e))?;

                let mut page = self
                    .inner
                    .get_instruments(params)
                    .await
                    .map_err(|e| anyhow::anyhow!(e))?;
                matched.append(&mut page);
                if !matched.is_empty() {
                    break;
                }
            }
            matched
        } else {
            let mut params = GetInstrumentsParamsBuilder::default();
            params.inst_type(instrument_type);
            params.inst_id(symbol);

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

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

        let raw_inst = resp
            .first()
            .ok_or_else(|| InstrumentLookupError::not_found(instrument_id))?;
        self.cache_trade_quote_ccy_list_from_instrument(raw_inst);

        // Skip pre-open instruments which have incomplete/empty field values
        if raw_inst.state == OKXInstrumentStatus::Preopen {
            return Err(OKXInstrumentDefinitionError::new(
                symbol,
                anyhow::anyhow!("instrument is in pre-open state"),
            )
            .into());
        }

        let ts_init = self.generate_ts_init();
        let Some(instrument) = parse_instrument_any(raw_inst, None, None, ts_init)
            .map_err(|e| OKXInstrumentDefinitionError::new(symbol, e))?
        else {
            return Err(OKXInstrumentDefinitionError::new(
                symbol,
                anyhow::anyhow!("unsupported instrument type"),
            )
            .into());
        };

        self.cache_instrument(instrument.clone());

        Ok(instrument)
    }

    async fn request_spread_instrument(&self, symbol: &str) -> anyhow::Result<InstrumentAny> {
        let resp = self
            .inner
            .get_spreads(GetSpreadsParams {
                sprd_id: Some(symbol.to_string()),
                ..Default::default()
            })
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        let raw_spread = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("Spread instrument {symbol} not found"))?;
        let ts_init = self.generate_ts_init();

        parse_spread_instrument(raw_spread, None, None, ts_init)
            .map_err(|e| OKXInstrumentDefinitionError::new(symbol, e).into())
    }

    /// Requests event contract series metadata from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the response cannot be deserialized.
    pub async fn request_event_contract_series(
        &self,
        params: GetEventContractSeriesParams,
    ) -> Result<Vec<OKXEventContractSeries>, OKXHttpError> {
        self.inner.get_event_contract_series(params).await
    }

    /// Requests event metadata for an event contract series from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the response cannot be deserialized.
    pub async fn request_event_contract_events(
        &self,
        params: GetEventContractEventsParams,
    ) -> Result<Vec<OKXEventContractEvent>, OKXHttpError> {
        self.inner.get_event_contract_events(params).await
    }

    /// Requests event contract market metadata from OKX.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the response cannot be deserialized.
    pub async fn request_event_contract_markets(
        &self,
        params: GetEventContractMarketsParams,
    ) -> Result<Vec<OKXEventContractMarket>, OKXHttpError> {
        self.inner.get_event_contract_markets(params).await
    }

    /// Requests the reference price for an OKX option-chain bootstrap.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or the option symbol is invalid.
    pub(crate) async fn request_option_chain_reference_price(
        &self,
        instrument_id: InstrumentId,
    ) -> anyhow::Result<Option<Price>> {
        let symbol = instrument_id.symbol.inner().as_str();
        let inst_family = extract_inst_family(symbol)?.to_string();
        let exp_time = Self::option_summary_exp_time(symbol)?;
        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 || summary.inst_id != symbol {
                continue;
            }

            let decimal = match Decimal::from_str(&summary.fwd_px) {
                Ok(decimal) if decimal > Decimal::ZERO => decimal,
                Ok(_) => return Ok(None),
                Err(e) => {
                    log::warn!("Invalid OKX option-chain reference price for {instrument_id}: {e}");
                    return Ok(None);
                }
            };
            return Price::from_decimal(decimal).map(Some).map_err(Into::into);
        }

        Ok(None)
    }

    /// 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 inst = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let mut params = GetMarkPriceParamsBuilder::default();
        params.inst_type(okx_instrument_type(&inst)?);
        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 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)
    }

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

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

        resp.first()
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("No price limit returned from OKX"))
    }

    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_by_id(instrument_id)?;
        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::debug!(
            "Fetched order book for {} with {} bids and {} asks",
            instrument_id,
            snapshot.bids.len(),
            snapshot.asks.len(),
        );

        Ok(book)
    }

    /// Requests an RPI order book snapshot for the `instrument_id`.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP request fails or book parsing fails.
    pub async fn request_rpi_book_snapshot(
        &self,
        instrument_id: InstrumentId,
        depth: Option<u32>,
    ) -> anyhow::Result<OrderBook> {
        let inst = self.instrument_from_cache_by_id(instrument_id)?;
        let price_precision = inst.price_precision();
        let size_precision = inst.size_precision();
        let params = GetRpiOrderBookParams {
            inst_id: instrument_id.symbol.to_string(),
            sz: depth,
        };
        let resp = self
            .inner
            .get_rpi_order_book(params)
            .await
            .map_err(|e| anyhow::anyhow!(e))?;
        let snapshot = resp
            .first()
            .ok_or_else(|| anyhow::anyhow!("No RPI 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 = Price::from_decimal_dp(level.0, price_precision)?;
            let size = Quantity::from_decimal_dp(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 = Price::from_decimal_dp(level.0, price_precision)?;
            let size = Quantity::from_decimal_dp(level.1, size_precision)?;
            let index = (bids_len + i) as u64;
            let order = BookOrder::new(OrderSide::Sell, price, size, index);
            book.add(order, 0, index, ts_event);
        }

        log::debug!(
            "Fetched RPI 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_by_id(instrument_id)?;
        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::debug!(
            "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<Timestamp>,
        end: Option<Timestamp>,
        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.as_millisecond().to_string());
        }

        if let Some(end) = end {
            params.after = Some(end.as_millisecond().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::debug!(
            "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<Timestamp>,
        end: Option<Timestamp>,
        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 = self.inner.clock.get_time_ns().to_datetime_utc();

        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(jiff::Timestamp::as_millisecond);
        let end_ms = end.map(jiff::Timestamp::as_millisecond);

        let ts_init = self.generate_ts_init();
        let inst = self.instrument_from_cache_by_id(instrument_id)?;

        // Historical pagination walks backwards using trade IDs, OKX does not honor 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;
                }

                time::sleep(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.
    /// Each bar uses the candle open time as `ts_event` and its close time as `ts_init`.
    /// Requests without an end may include an unconfirmed candle with a future `ts_init`.
    /// With bounds, candle open times are at or after `start` and close times are at or before `end`.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    ///
    /// # Endpoint Selection
    ///
    /// Normal pagination selects between endpoints with different limits:
    /// - Regular endpoint (`/api/v5/market/candles`): ≤ 300 rows/call, ≤ 40 req/2s
    ///   - Used when the pagination cursor is less than two hours old
    /// - History endpoint (`/api/v5/market/history-candles`): ≤ 100 rows/call, ≤ 20 req/2s
    ///   - Used when the pagination cursor is at least two hours old
    ///
    /// Cursor age uses the clock time captured when the request starts. Empty-page fallbacks
    /// can select a different endpoint.
    ///
    /// # 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
    ///
    /// - Range requests use exclusive `after` and `before` cursors, adjusting `before` to include
    ///   the floored start boundary
    /// - 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<Timestamp>,
        mut end: Option<Timestamp>,
        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 = self.inner.clock.get_time_ns().to_datetime_utc();

        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| i64::try_from(s.as_nanosecond()).ok());
        let end_ns = end.and_then(|e| i64::try_from(e.as_nanosecond()).ok());

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

            if slot_ms > 0 {
                (ms / slot_ms) * slot_ms // Floor to nearest bar boundary
            } else {
                ms
            }
        });

        let end_ms = end.map(Timestamp::as_millisecond);
        let now_ms = now.as_millisecond();

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

        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.map(|v| v.saturating_sub(1)),
            Mode::Latest => None,
        };

        // For Range mode, we'll paginate backwards like Backward mode
        let 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;
                    }
                }

                // 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 mut page: Vec<Bar> = Vec::with_capacity(raw.len());
            for r in &raw {
                page.push(parse_historical_candlestick(
                    r,
                    bar_type,
                    inst.price_precision(),
                    inst.size_precision(),
                )?);
            }
            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);

            let mut filtered: Vec<Bar> = 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| b.ts_init.as_i64() <= 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);
                        }
                    }
                }
            }

            if contribution == 0 {
                // An unaligned end can exclude the newest candle by its close time
                match mode {
                    Mode::Range => after_ms = page_oldest_ms,
                    Mode::Backward => before_ms = page_oldest_ms,
                    Mode::Latest => {
                        if let Some(b) = before_ms {
                            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;
                }
            } 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 {
                                after_ms = Some(oldest);
                            } 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
            {
                break;
            }

            time::sleep(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 mut page: Vec<Bar> = Vec::with_capacity(raw.len());
                for r in &raw {
                    page.push(parse_historical_candlestick(
                        r,
                        bar_type,
                        inst.price_precision(),
                        inst.size_precision(),
                    )?);
                }
                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| b.ts_init.as_i64() <= 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<Timestamp>,
        end: Option<Timestamp>,
        open_only: bool,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        Ok(self
            .request_order_status_reports_scoped(
                account_id,
                instrument_type,
                instrument_id,
                start,
                end,
                open_only,
                limit,
                None,
                None,
            )
            .await?
            .reports)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn request_order_status_reports_scoped(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        open_only: bool,
        limit: Option<u32>,
        scope: Option<ReportInstrumentScope<'_>>,
        max_history_pages: Option<u32>,
    ) -> anyhow::Result<ReportSweep<OrderStatusReport>> {
        if instrument_id.as_ref().is_some_and(|id| self.is_spread(*id))
            || (instrument_id.is_none() && instrument_type.is_none())
        {
            return self
                .request_spread_order_status_reports_scoped(
                    account_id,
                    instrument_id,
                    start,
                    end,
                    open_only,
                    limit,
                    scope,
                    max_history_pages,
                )
                .await;
        }

        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, mut complete) = if open_only {
            let pending = self.paginate_orders_pending(&pending_base, limit).await?;
            (pending.items, pending.complete)
        } else {
            let (history, pending) = Box::pin(async {
                tokio::try_join!(
                    self.paginate_orders_history(
                        &history_base,
                        limit,
                        reconciliation_history_page_limit(max_history_pages),
                    ),
                    self.paginate_orders_pending(&pending_base, limit),
                )
            })
            .await?;
            let mut combined_resp = history.items;
            combined_resp.extend(pending.items);
            (combined_resp, history.complete && pending.complete)
        };

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

            // Open orders are authoritative regardless of age; only closed
            // history respects the report window.
            if report_ts_outside_window(order.u_time, start_ns, end_ns)
                && !is_open_okx_order(order.state)
            {
                continue;
            }

            let inst = match self.resolve_report_instrument(
                order.inst_id,
                order.inst_type,
                false,
                is_open_okx_order(order.state),
                scope,
            )? {
                InstrumentResolution::Found(inst) => inst,
                InstrumentResolution::Skip => continue,
                InstrumentResolution::Incomplete => {
                    complete = false;
                    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::warn!("Failed to parse order status report: {e}");
                    complete = false;
                    continue;
                }
            };

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

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

            reports.push(report);
        }

        Ok(ReportSweep { reports, complete })
    }

    /// Requests a regular order status report by client order identifier.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the report cannot be parsed.
    pub async fn request_order_status_report(
        &self,
        account_id: AccountId,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        self.request_order_status_report_by_identifier(
            account_id,
            instrument_id,
            Some(client_order_id),
            None,
        )
        .await
    }

    pub(crate) async fn request_order_status_report_by_venue_order_id(
        &self,
        account_id: AccountId,
        instrument_id: InstrumentId,
        venue_order_id: VenueOrderId,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        self.request_order_status_report_by_identifier(
            account_id,
            instrument_id,
            None,
            Some(venue_order_id),
        )
        .await
    }

    async fn request_order_status_report_by_identifier(
        &self,
        account_id: AccountId,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let mut params_builder = GetOrderParamsBuilder::default();
        params_builder.inst_id(instrument_id.symbol.inner().to_string());

        match (client_order_id, venue_order_id) {
            (Some(client_order_id), None) => {
                params_builder.cl_ord_id(client_order_id.as_str().to_string());
            }
            (None, Some(venue_order_id)) => {
                params_builder.ord_id(venue_order_id.as_str().to_string());
            }
            _ => anyhow::bail!(
                "Exactly one of client_order_id or venue_order_id is required for an order detail request"
            ),
        }

        let params = params_builder
            .build()
            .map_err(|e| anyhow::anyhow!("Failed to build order detail params: {e}"))?;
        let orders = match self.inner.get_order(params).await {
            Ok(orders) => orders,
            Err(e) if e.is_order_not_found() => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        let order = match orders.as_slice() {
            [] => return Ok(None),
            [order] => order,
            _ => anyhow::bail!(
                "Order detail returned {} records for one identifier",
                orders.len(),
            ),
        };

        if order.inst_id.as_str() != instrument_id.symbol.inner() {
            anyhow::bail!(
                "Order detail instrument mismatch for {instrument_id}: returned {}",
                order.inst_id,
            );
        }

        if let Some(venue_order_id) = venue_order_id
            && order.ord_id.as_str() != venue_order_id.as_str()
        {
            anyhow::bail!(
                "Order detail venue order ID mismatch for {venue_order_id}: returned {}",
                order.ord_id,
            );
        }

        let ts_init = self.generate_ts_init();
        let mut report = parse_order_status_report(
            order,
            account_id,
            instrument.id(),
            instrument.price_precision(),
            instrument.size_precision(),
            ts_init,
        )?;
        let post_only_rejected = order.state == OKXOrderStatus::Canceled
            && report.filled_qty.is_zero()
            && (order.cancel_source == OKX_POST_ONLY_CANCEL_SOURCE
                || order.cancel_source_reason.contains("POST_ONLY"));
        if post_only_rejected {
            report.order_status = OrderStatus::Rejected;
            report.post_only = true;
            let reason = if order.cancel_source_reason.is_empty() {
                OKX_POST_ONLY_CANCEL_REASON.to_string()
            } else {
                order.cancel_source_reason.clone()
            };
            report.cancel_reason = Some(reason);
        }

        Ok(Some(report))
    }

    /// Requests spread order status reports for the given parameters.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn request_spread_order_status_reports(
        &self,
        account_id: AccountId,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        open_only: bool,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        Ok(self
            .request_spread_order_status_reports_scoped(
                account_id,
                instrument_id,
                start,
                end,
                open_only,
                limit,
                None,
                None,
            )
            .await?
            .reports)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn request_spread_order_status_reports_scoped(
        &self,
        account_id: AccountId,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        open_only: bool,
        limit: Option<u32>,
        scope: Option<ReportInstrumentScope<'_>>,
        max_history_pages: Option<u32>,
    ) -> anyhow::Result<ReportSweep<OrderStatusReport>> {
        let mut pending_builder = GetSpreadOrdersParamsBuilder::default();
        let mut history_builder = GetSpreadOrdersParamsBuilder::default();

        if let Some(instrument_id) = instrument_id.as_ref() {
            let sprd_id = instrument_id.symbol.inner().to_string();
            pending_builder.sprd_id(sprd_id.clone());
            history_builder.sprd_id(sprd_id);
        }

        if let Some(start) = start {
            history_builder.begin(start.as_millisecond().to_string());
        }

        if let Some(end) = end {
            history_builder.end(end.as_millisecond().to_string());
        }

        if let Some(limit) = spread_page_limit(limit) {
            pending_builder.limit(limit);
            history_builder.limit(limit);
        }

        let pending_base = pending_builder.build().map_err(|e| anyhow::anyhow!(e))?;
        let history_base = history_builder.build().map_err(|e| anyhow::anyhow!(e))?;

        let (combined_resp, mut complete) = if open_only {
            let pending = self
                .paginate_spread_orders_pending(&pending_base, limit)
                .await?;
            (pending.items, pending.complete)
        } else {
            let (history, pending) = Box::pin(async {
                tokio::try_join!(
                    self.paginate_spread_orders_history(
                        &history_base,
                        limit,
                        reconciliation_history_page_limit(max_history_pages),
                    ),
                    self.paginate_spread_orders_pending(&pending_base, limit),
                )
            })
            .await?;
            let mut combined_resp = history.items;
            combined_resp.extend(pending.items);
            (combined_resp, history.complete && pending.complete)
        };

        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());
        let mut seen: AHashSet<String> = AHashSet::new();

        for order in combined_resp {
            let seen_key = if order.cl_ord_id.is_empty() {
                order.ord_id.as_str().to_string()
            } else {
                order.cl_ord_id.as_str().to_string()
            };

            if !seen.insert(seen_key) {
                continue;
            }

            // Open spread orders are authoritative regardless of age; only
            // closed history respects the report window.
            if let Some(ts) = order.u_time.or(order.c_time)
                && report_ts_outside_window(ts, start_ns, end_ns)
                && !is_open_okx_order(order.state)
            {
                continue;
            }

            let inst = match self.resolve_report_instrument(
                order.sprd_id,
                OKXInstrumentType::Any,
                true,
                is_open_okx_order(order.state),
                scope,
            )? {
                InstrumentResolution::Found(inst) => inst,
                InstrumentResolution::Skip => continue,
                InstrumentResolution::Incomplete => {
                    complete = false;
                    continue;
                }
            };

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

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

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

            reports.push(report);
        }

        Ok(ReportSweep { reports, complete })
    }

    async fn paginate_spread_orders_history(
        &self,
        base: &GetSpreadOrdersParams,
        limit: Option<u32>,
        max_pages: usize,
    ) -> anyhow::Result<PageSweep<OKXSpreadOrder>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

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

            let page = self
                .inner
                .get_spread_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!(
                "Spread order history pagination hit {max_pages} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

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

        Ok(PageSweep::from_pages(all, exhausted))
    }

    async fn paginate_spread_orders_pending(
        &self,
        base: &GetSpreadOrdersParams,
        limit: Option<u32>,
    ) -> anyhow::Result<PageSweep<OKXSpreadOrder>> {
        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.end_id = cursor.take();

            let page = self
                .inner
                .get_spread_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 spread 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(PageSweep::from_pages(all, exhausted))
    }

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

        for _ in 0..max_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_pages} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

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

        Ok(PageSweep::from_pages(all, exhausted))
    }

    // Paginates through pending orders using `ord_id` as the cursor
    async fn paginate_orders_pending(
        &self,
        base: &GetOrderListParams,
        limit: Option<u32>,
    ) -> anyhow::Result<PageSweep<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(PageSweep::from_pages(all, exhausted))
    }

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

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

            let page = match history {
                FillHistory::Recent => self.inner.get_fills(params).await,
                FillHistory::Extended => self.inner.get_fills_history(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_pages} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

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

        Ok(PageSweep::from_pages(all, exhausted))
    }

    // Paginates through pending algo orders using `algo_id` as the cursor
    async fn paginate_algo_pending(
        &self,
        base: &GetAlgoOrdersParams,
        limit: Option<usize>,
        require_complete_active_coverage: bool,
    ) -> anyhow::Result<PageSweep<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 && !require_complete_active_coverage =>
                {
                    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(PageSweep::from_pages(all, exhausted))
    }

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

        for _ in 0..max_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 && !require_complete_active_coverage =>
                {
                    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_pages} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

        Ok(PageSweep::from_pages(all, exhausted))
    }

    /// 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<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<FillReport>> {
        Ok(self
            .request_fill_reports_scoped(
                account_id,
                instrument_type,
                instrument_id,
                start,
                end,
                limit,
                FillHistory::Recent,
                None,
                None,
            )
            .await?
            .reports)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn request_fill_reports_scoped(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<u32>,
        history: FillHistory,
        scope: Option<ReportInstrumentScope<'_>>,
        max_history_pages: Option<u32>,
    ) -> anyhow::Result<ReportSweep<FillReport>> {
        if instrument_id.as_ref().is_some_and(|id| self.is_spread(*id))
            || (instrument_id.is_none() && instrument_type.is_none())
        {
            return self
                .request_spread_fill_reports_scoped(
                    account_id,
                    instrument_id,
                    start,
                    end,
                    limit,
                    scope,
                    max_history_pages,
                )
                .await;
        }

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

        if let Some(start) = start {
            params.begin(start.as_millisecond().to_string());
        }

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

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

        let sweep = self
            .paginate_fills(
                &params,
                limit,
                history,
                reconciliation_history_page_limit(max_history_pages),
            )
            .await?;
        let mut complete = sweep.complete;

        // 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(sweep.items.len());

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

            if !fill_quantity_is_positive(&detail.fill_sz).with_context(|| {
                format!(
                    "failed to parse fill quantity for instrument {}",
                    detail.inst_id
                )
            })? {
                continue;
            }

            if report_ts_outside_window(detail.ts, start_ns, end_ns) {
                continue;
            }

            let inst = match self.resolve_report_instrument(
                detail.inst_id,
                detail.inst_type,
                false,
                false,
                scope,
            )? {
                InstrumentResolution::Found(inst) => inst,
                InstrumentResolution::Skip => continue,
                InstrumentResolution::Incomplete => {
                    complete = false;
                    continue;
                }
            };

            let report = parse_fill_report(
                &detail,
                account_id,
                inst.id(),
                inst.price_precision(),
                inst.size_precision(),
                ts_init,
            )
            .with_context(|| {
                format!(
                    "failed to parse fill report for instrument {}",
                    detail.inst_id
                )
            })?;

            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(ReportSweep { reports, complete })
    }

    /// Requests spread fill reports for the given parameters.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn request_spread_fill_reports(
        &self,
        account_id: AccountId,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<FillReport>> {
        Ok(self
            .request_spread_fill_reports_scoped(
                account_id,
                instrument_id,
                start,
                end,
                limit,
                None,
                None,
            )
            .await?
            .reports)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn request_spread_fill_reports_scoped(
        &self,
        account_id: AccountId,
        instrument_id: Option<InstrumentId>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        limit: Option<u32>,
        scope: Option<ReportInstrumentScope<'_>>,
        max_history_pages: Option<u32>,
    ) -> anyhow::Result<ReportSweep<FillReport>> {
        let mut builder = GetSpreadTradesParamsBuilder::default();

        if let Some(instrument_id) = instrument_id.as_ref() {
            builder.sprd_id(instrument_id.symbol.inner().to_string());
        }

        if let Some(start) = start {
            builder.begin(start.as_millisecond().to_string());
        }

        if let Some(end) = end {
            builder.end(end.as_millisecond().to_string());
        }

        if let Some(limit) = spread_page_limit(limit) {
            builder.limit(limit);
        }

        let params = builder.build().map_err(|e| anyhow::anyhow!(e))?;
        let sweep = self
            .paginate_spread_fills(
                &params,
                limit,
                reconciliation_history_page_limit(max_history_pages),
            )
            .await?;
        let mut complete = sweep.complete;

        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(sweep.items.len());

        for detail in sweep.items {
            if detail.fill_sz.is_empty() {
                continue;
            }

            if !fill_quantity_is_positive(&detail.fill_sz).with_context(|| {
                format!(
                    "failed to parse spread fill quantity for instrument {}",
                    detail.sprd_id
                )
            })? {
                continue;
            }

            if report_ts_outside_window(detail.ts, start_ns, end_ns) {
                continue;
            }

            let inst = match self.resolve_report_instrument(
                detail.sprd_id,
                OKXInstrumentType::Any,
                true,
                false,
                scope,
            )? {
                InstrumentResolution::Found(inst) => inst,
                InstrumentResolution::Skip => continue,
                InstrumentResolution::Incomplete => {
                    complete = false;
                    continue;
                }
            };

            let report = parse_spread_fill_report(
                &detail,
                account_id,
                inst.id(),
                inst.price_precision(),
                inst.size_precision(),
                ts_init,
            )
            .with_context(|| {
                format!(
                    "failed to parse spread fill report for instrument {}",
                    detail.sprd_id
                )
            })?;

            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(ReportSweep { reports, complete })
    }

    async fn paginate_spread_fills(
        &self,
        base: &GetSpreadTradesParams,
        limit: Option<u32>,
        max_pages: usize,
    ) -> anyhow::Result<PageSweep<OKXSpreadTrade>> {
        let mut all = Vec::new();
        let mut cursor: Option<String> = None;
        let mut exhausted = true;

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

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

            let page_len = page.len();
            cursor = page.last().map(|o| o.trade_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!(
                "Spread fill pagination hit {max_pages} page cap, \
                 results may be truncated ({} records)",
                all.len()
            );
        }

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

        Ok(PageSweep::from_pages(all, exhausted))
    }

    /// 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, a position cannot be parsed, or multiple
    /// nonzero positions map to the same hedge leg.
    ///
    /// # 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>> {
        Ok(self
            .request_position_status_reports_scoped(
                account_id,
                instrument_type,
                instrument_id,
                None,
            )
            .await?
            .reports)
    }

    pub(crate) async fn request_position_status_reports_scoped(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        scope: Option<ReportInstrumentScope<'_>>,
    ) -> anyhow::Result<ReportSweep<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<PositionStatusReport> = Vec::with_capacity(resp.len());
        let mut position_ids = AHashMap::new();

        for position in resp {
            let inst = match self.resolve_report_instrument(
                position.inst_id,
                position.inst_type,
                false,
                true,
                scope,
            )? {
                InstrumentResolution::Found(inst) => inst,
                InstrumentResolution::Skip => continue,
                InstrumentResolution::Incomplete => {
                    anyhow::bail!(
                        "Instrument {} missing from cache for position report",
                        position.inst_id
                    );
                }
            };

            if let Some(scope) = scope {
                scope.validate_position_margin(&position)?;
            }

            let report = parse_position_status_report(
                &position,
                account_id,
                inst.id(),
                inst.size_precision(),
                ts_init,
            )
            .with_context(|| {
                format!(
                    "failed to parse position status report for instrument {}",
                    position.inst_id
                )
            })?;

            if let Some(position_id) = report.venue_position_id {
                if let Some(index) = position_ids.get(&position_id).copied() {
                    if report.quantity.is_zero() {
                        continue;
                    }

                    let existing: &mut PositionStatusReport = &mut reports[index];
                    if existing.quantity.is_zero() {
                        *existing = report;
                        continue;
                    }

                    anyhow::bail!(
                        "Multiple OKX positions map to {position_id}; mixed margin modes are not supported"
                    );
                }

                position_ids.insert(position_id, reports.len());
            }

            reports.push(report);
        }

        Ok(ReportSweep {
            reports,
            complete: true,
        })
    }

    /// 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 = base.code;
                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)?;

        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(OKXHttpError::EmptyResponse)
    }

    /// Places multiple regular orders via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn place_orders(
        &self,
        requests: Vec<OKXPlaceOrderRequest>,
    ) -> Result<Vec<OKXPlaceOrderResponse>, OKXHttpError> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let body = serde_json::to_vec(&requests)?;
        self.inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/batch-orders",
                None,
                Some(body),
                true,
            )
            .await
    }

    /// Amends a regular order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn amend_order(
        &self,
        request: OKXAmendOrderRequest,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        let body = serde_json::to_vec(&request)?;
        let resp: Vec<OKXPlaceOrderResponse> = self
            .inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/amend-order",
                None,
                Some(body),
                true,
            )
            .await?;

        resp.into_iter().next().ok_or(OKXHttpError::EmptyResponse)
    }

    /// Amends multiple regular orders via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn amend_orders(
        &self,
        requests: Vec<OKXAmendOrderRequest>,
    ) -> Result<Vec<OKXPlaceOrderResponse>, OKXHttpError> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let body = serde_json::to_vec(&requests)?;
        self.inner
            .send_request::<_, ()>(
                Method::POST,
                "/api/v5/trade/amend-batch-orders",
                None,
                Some(body),
                true,
            )
            .await
    }

    /// Places a spread order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn place_spread_order(
        &self,
        request: OKXPlaceSpreadOrderRequest,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        let resp = self.inner.place_spread_order(request).await?;
        let item = resp.into_iter().next().ok_or(OKXHttpError::EmptyResponse)?;

        if let Some(ref code) = item.s_code
            && code != OKX_SUCCESS_CODE
        {
            let msg = item.s_msg.clone().unwrap_or_default();
            return Err(OKXHttpError::from_venue_response(code.clone(), msg, None));
        }

        Ok(item)
    }

    /// Cancels a spread order via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn cancel_spread_order(
        &self,
        request: OKXCancelSpreadOrderRequest,
    ) -> Result<OKXCancelOrderResponse, OKXHttpError> {
        let resp = self.inner.cancel_spread_order(request).await?;
        let item = resp.into_iter().next().ok_or(OKXHttpError::EmptyResponse)?;

        if let Some(ref code) = item.s_code
            && code != OKX_SUCCESS_CODE
        {
            let msg = item.s_msg.clone().unwrap_or_default();
            return Err(OKXHttpError::from_venue_response(code.clone(), msg, None));
        }

        Ok(item)
    }

    /// Cancels all orders for a spread via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn cancel_all_spread_orders(
        &self,
        spread_id: InstrumentId,
    ) -> Result<Vec<OKXCancelOrderResponse>, OKXHttpError> {
        let request = OKXCancelAllSpreadOrdersRequest {
            sprd_id: spread_id.symbol.as_str().to_string(),
        };

        self.inner.cancel_all_spread_orders(request).await
    }

    /// Cancels an order via HTTP, routing spread instruments to the spread endpoint.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or if no order identifier is supplied.
    pub async fn cancel_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> Result<OKXCancelOrderResponse, OKXHttpError> {
        if client_order_id.is_none() && venue_order_id.is_none() {
            return Err(OKXHttpError::ValidationError(
                "Either `client_order_id` or `venue_order_id` is required".to_string(),
            ));
        }

        let ord_id = venue_order_id.as_ref().map(ToString::to_string);
        let cl_ord_id = if venue_order_id.is_none() {
            client_order_id.as_ref().map(|id| id.as_str().to_string())
        } else {
            None
        };

        if self.is_spread(instrument_id) {
            return self
                .cancel_spread_order(OKXCancelSpreadOrderRequest { ord_id, cl_ord_id })
                .await;
        }

        let request = OKXCancelOrderRequest {
            inst_id: instrument_id.symbol.as_str().to_string(),
            inst_id_code: None,
            ord_id,
            cl_ord_id,
        };
        let mut resp = self.cancel_orders(vec![request]).await?;
        let item = resp.pop().ok_or(OKXHttpError::EmptyResponse)?;

        if let Some(ref code) = item.s_code
            && code != OKX_SUCCESS_CODE
        {
            let msg = item.s_msg.clone().unwrap_or_default();
            return Err(OKXHttpError::from_venue_response(code.clone(), msg, None));
        }

        Ok(item)
    }

    /// Cancels all open orders for an instrument via HTTP.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails.
    pub async fn cancel_all_orders(
        &self,
        instrument_id: InstrumentId,
    ) -> Result<Vec<OKXCancelOrderResponse>, OKXHttpError> {
        if self.is_spread(instrument_id) {
            return self.cancel_all_spread_orders(instrument_id).await;
        }

        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()))?;

        let mut pending_base = GetOrderListParamsBuilder::default();
        pending_base.inst_type(instrument_type);
        pending_base.inst_id(instrument_id.symbol.inner().to_string());
        let pending_base = pending_base
            .build()
            .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;

        let pending = self
            .paginate_orders_pending(&pending_base, None)
            .await
            .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;
        let requests = pending
            .items
            .into_iter()
            .map(|order| OKXCancelOrderRequest {
                inst_id: order.inst_id.to_string(),
                inst_id_code: None,
                ord_id: if order.ord_id.is_empty() {
                    None
                } else {
                    Some(order.ord_id.to_string())
                },
                cl_ord_id: if order.ord_id.is_empty() && !order.cl_ord_id.is_empty() {
                    Some(order.cl_ord_id.to_string())
                } else {
                    None
                },
            })
            .collect();

        self.cancel_orders(requests).await
    }

    /// Cancels multiple regular 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-trade-post-cancel-multiple-orders>
    pub async fn cancel_orders(
        &self,
        requests: Vec<OKXCancelOrderRequest>,
    ) -> Result<Vec<OKXCancelOrderResponse>, OKXHttpError> {
        if requests.is_empty() {
            return Ok(Vec::new());
        }

        let body = serde_json::to_vec(&requests)?;

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

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

        Ok(resp)
    }

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

        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(OKXHttpError::EmptyResponse)?;

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

        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])?;

        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(OKXHttpError::EmptyResponse)?;

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

        Ok(item)
    }

    /// Cancels multiple algo orders via HTTP in a single request.
    ///
    /// Items with non-zero `sCode` are logged at debug level 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)?;

        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::debug!(
                    "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 at debug level.
    ///
    /// # 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)?;

        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::debug!(
                    "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)?;

        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(OKXHttpError::EmptyResponse)
    }

    /// 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_sl_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>,
        new_tp_trigger_price: Option<Price>,
        new_tp_order_price: Option<String>,
        new_tp_trigger_px_type: Option<String>,
        new_sl_order_price: Option<String>,
        new_sl_trigger_px_type: Option<String>,
    ) -> 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_trigger_px_type: None,
            new_tp_trigger_px: new_tp_trigger_price.map(|p| p.to_string()),
            new_tp_ord_px: new_tp_order_price,
            new_tp_trigger_px_type,
            new_sl_trigger_px: new_sl_trigger_price.map(|p| p.to_string()),
            new_sl_ord_px: new_sl_order_price,
            new_sl_trigger_px_type,
            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>,
        outcome: Option<String>,
        slippage_pct: Option<String>,
        rpi: Option<bool>,
        rpi_taker_access: Option<bool>,
        rpi_px_round: Option<bool>,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        let rpi = rpi.unwrap_or(false);

        if self.is_spread(instrument_id) {
            if reduce_only.unwrap_or(false)
                || quote_quantity.unwrap_or(false)
                || attach_algo_ords
                    .as_ref()
                    .is_some_and(|orders| !orders.is_empty())
                || px_usd.is_some()
                || px_vol.is_some()
                || outcome.is_some()
                || slippage_pct.is_some()
                || rpi
                || rpi_taker_access.is_some()
                || rpi_px_round.is_some()
            {
                return Err(OKXHttpError::ValidationError(
                    "OKX spread orders do not support regular order extensions".to_string(),
                ));
            }

            return self
                .place_spread_order_with_domain_types(
                    instrument_id,
                    client_order_id,
                    order_side,
                    order_type,
                    quantity,
                    time_in_force,
                    price,
                    post_only,
                )
                .await;
        }

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

        if rpi && order_type != OrderType::Limit {
            return Err(OKXHttpError::ValidationError(
                "OKX RPI orders require a limit order".to_string(),
            ));
        }

        let (ord_type, px) = if rpi {
            (OKXOrderType::Rpi, price)
        } else 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)
        };

        if instrument_type == OKXInstrumentType::Events && outcome.is_none() {
            return Err(OKXHttpError::ValidationError(
                "OKX event contract orders require `outcome`".to_string(),
            ));
        }

        let reduce_only = if reduce_only == Some(true) || position_side.is_some() {
            okx_reduce_only_wire_value(
                instrument_type,
                td_mode,
                order_side,
                position_side,
                reduce_only,
            )
            .map_err(OKXHttpError::ValidationError)?
        } 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 trade_quote_ccy =
            self.resolve_spot_trade_quote_ccy(instrument_type, instrument_id.symbol.inner())?;

        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,
            trade_quote_ccy,
            attach_algo_ords,
            outcome,
            slippage_pct,
            rpi_taker_access,
            rpi_px_round,
        };

        self.place_order(request).await
    }

    /// Places a spread order using domain types.
    ///
    /// # Errors
    ///
    /// Returns an error if the order cannot be represented by the OKX spread endpoint
    /// or if the request fails.
    #[expect(clippy::too_many_arguments)]
    pub async fn place_spread_order_with_domain_types(
        &self,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: Option<TimeInForce>,
        price: Option<Price>,
        post_only: Option<bool>,
    ) -> Result<OKXPlaceOrderResponse, OKXHttpError> {
        if !self.is_spread(instrument_id) {
            return Err(OKXHttpError::ValidationError(format!(
                "Instrument is not an OKX spread: {instrument_id}",
            )));
        }

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

        if !matches!(order_type, OrderType::Limit) {
            return Err(OKXHttpError::ValidationError(
                "OKX spread orders support Limit orders only".to_string(),
            ));
        }

        let ord_type = if post_only.unwrap_or(false) {
            OKXOrderType::PostOnly
        } else if matches!(time_in_force, Some(TimeInForce::Ioc)) {
            OKXOrderType::Ioc
        } else if matches!(time_in_force, Some(TimeInForce::Fok)) {
            return Err(OKXHttpError::ValidationError(
                "OKX spread orders do not support FOK time-in-force".to_string(),
            ));
        } else {
            OKXOrderType::Limit
        };

        let price = price.ok_or_else(|| {
            OKXHttpError::ValidationError("OKX spread orders require a limit price".to_string())
        })?;

        let request = OKXPlaceSpreadOrderRequest {
            sprd_id: instrument_id.symbol.as_str().to_string(),
            cl_ord_id: Some(client_order_id.as_str().to_string()),
            tag: Some(OKX_NAUTILUS_BROKER_ID.to_string()),
            side: OKXSide::from(order_side),
            ord_type,
            sz: quantity.to_string(),
            px: Some(price.to_string()),
        };

        self.place_spread_order(request).await
    }

    /// Places an algo order using domain types.
    ///
    /// Spot `StopMarket` and `MarketIfTouched` buys use `conditional` orders with explicit
    /// base (`quote_quantity=false`) or quote (`true`) sizing. Spot trailing buys require
    /// quote sizing. Other algo orders require base size or derivative contracts.
    /// Cash-spot triggers support last price only; cross/isolated feeds depend on the account mode
    /// and algo type.
    /// Trailing orders reject mark/index feeds.
    ///
    /// # Errors
    ///
    /// Returns an error if the quantity denomination or trigger feed is unsupported,
    /// required trigger terms are missing, or 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,
        quote_quantity: bool,
        trigger_price: Option<Price>,
        trigger_type: Option<TriggerType>,
        limit_price: Option<Price>,
        reduce_only: Option<bool>,
        position_side: Option<PositionSide>,
        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);

        let instrument_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());

        let tgt_ccy = okx_algo_target_currency(
            instrument_type,
            td_mode,
            order_side,
            order_type,
            quote_quantity,
        )
        .map_err(OKXHttpError::from_order_denial)?;

        validate_algo_trigger_type(instrument_type, td_mode, order_type, trigger_type)
            .map_err(OKXHttpError::ValidationError)?;

        let ccy = if instrument_type == OKXInstrumentType::Spot && td_mode != OKXTradeMode::Cash {
            let (_, quote) = parse_base_quote_from_symbol(instrument_id.symbol.as_str())
                .map_err(|e| OKXHttpError::ValidationError(e.to_string()))?;
            Some(quote.to_string())
        } else {
            None
        };

        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 || tgt_ccy.is_some() {
            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 conditional 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,
                (!uses_close_fraction).then(|| quantity.to_string()),
                None,
                None,
                None,
                sl_trigger_px,
                sl_ord_px,
                sl_trigger_px_type,
                tp_trigger_px,
                tp_ord_px,
                tp_trigger_px_type,
                if uses_close_fraction {
                    Some(position_side.map_or(OKXPositionSide::Net, Into::into))
                } else {
                    position_side.map(Into::into)
                },
                if uses_close_fraction {
                    Some(true)
                } else {
                    reduce_only
                },
            )
        } 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,
                position_side.map(Into::into),
                reduce_only,
            )
        };

        let reduce_only = if position_side.is_some() {
            okx_reduce_only_wire_value(
                instrument_type,
                td_mode,
                order_side,
                position_side,
                reduce_only,
            )
            .map_err(OKXHttpError::ValidationError)?
        } else {
            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,
            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()),
            ccy,
        };

        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<OKXAlgoOrderStatus>,
        limit: Option<u32>,
    ) -> anyhow::Result<Vec<OrderStatusReport>> {
        Ok(self
            .request_algo_order_status_reports_sweep(
                account_id,
                instrument_type,
                instrument_id,
                algo_id,
                algo_client_order_id,
                state,
                limit,
                None,
                None,
                false,
                None,
            )
            .await?
            .reports)
    }

    #[expect(clippy::too_many_arguments)]
    pub(crate) async fn request_algo_order_status_reports_sweep(
        &self,
        account_id: AccountId,
        instrument_type: Option<OKXInstrumentType>,
        instrument_id: Option<InstrumentId>,
        algo_id: Option<String>,
        algo_client_order_id: Option<ClientOrderId>,
        state: Option<OKXAlgoOrderStatus>,
        limit: Option<u32>,
        start: Option<Timestamp>,
        end: Option<Timestamp>,
        require_complete_active_coverage: bool,
        max_history_pages: Option<u32>,
    ) -> anyhow::Result<AlgoOrderReportSweep> {
        let mut instruments_cache: AHashMap<Ustr, InstrumentAny> = AHashMap::new();
        let mut ambiguous_triggered_child_ids = AHashSet::new();
        let has_specific_lookup = algo_id.is_some() || algo_client_order_id.is_some();
        let start_ns = start.map(UnixNanos::from);
        let end_ns = end.map(UnixNanos::from);
        let mut complete = true;

        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: AHashMap<(String, String), usize> = AHashMap::new();

        if has_specific_lookup {
            let mut params_builder = GetAlgoOrderParamsBuilder::default();

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

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

            let params = params_builder
                .build()
                .map_err(|e| anyhow::anyhow!(format!("Failed to build algo order params: {e}")))?;
            let mut orders = match self.inner.get_algo_order(params).await {
                Ok(orders) => orders,
                Err(e) if e.is_order_not_found() => {
                    return Ok(AlgoOrderReportSweep {
                        reports,
                        complete,
                        ambiguous_triggered_child_ids,
                    });
                }
                Err(e) => return Err(e.into()),
            };

            if let Some(state) = state {
                orders.retain(|order| order.state == state);
            }

            self.resolve_close_fraction_sizes(&mut orders).await?;

            complete &= self.collect_algo_reports(
                account_id,
                &orders,
                false,
                &mut instruments_cache,
                ts_init,
                start_ns,
                end_ns,
                &mut seen,
                &mut reports,
                &mut ambiguous_triggered_child_ids,
            )?;

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

            return Ok(AlgoOrderReportSweep {
                reports,
                complete,
                ambiguous_triggered_child_ids,
            });
        }

        let query_pending = state.is_none()
            || matches!(
                state,
                Some(OKXAlgoOrderStatus::Live | OKXAlgoOrderStatus::Pause)
            );
        let history_states: &[OKXAlgoOrderStatus] = match state {
            None => &[
                OKXAlgoOrderStatus::Effective,
                OKXAlgoOrderStatus::Canceled,
                OKXAlgoOrderStatus::OrderFailed,
            ],
            // An unrecognized state cannot be queried by name and matches nothing
            Some(OKXAlgoOrderStatus::Unknown) => &[],
            Some(OKXAlgoOrderStatus::Live | OKXAlgoOrderStatus::Pause) => &[],
            Some(
                OKXAlgoOrderStatus::Effective
                | OKXAlgoOrderStatus::OrderPlaced
                | OKXAlgoOrderStatus::PartiallyEffective
                | OKXAlgoOrderStatus::Filled,
            ) => &[OKXAlgoOrderStatus::Effective],
            Some(OKXAlgoOrderStatus::Canceled) => &[OKXAlgoOrderStatus::Canceled],
            Some(OKXAlgoOrderStatus::OrderFailed | OKXAlgoOrderStatus::PartiallyFailed) => {
                &[OKXAlgoOrderStatus::OrderFailed]
            }
        };

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

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

            if query_pending {
                let remaining = limit.map(|l| (l as usize).saturating_sub(reports.len()));
                let pending_sweep = match self
                    .paginate_algo_pending(&params, remaining, require_complete_active_coverage)
                    .await
                {
                    Ok(sweep) => sweep,
                    Err(e) if require_complete_active_coverage => {
                        return Err(OKXPendingAlgoOrderReportsError::new(e).into());
                    }
                    Err(e) => return Err(e),
                };

                if require_complete_active_coverage && !pending_sweep.complete {
                    return Err(OKXPendingAlgoOrderReportsError::new(anyhow::anyhow!(
                        "Pending {ord_type:?} algo order pagination for {inst_type:?} did not establish complete coverage"
                    ))
                    .into());
                }
                complete &= pending_sweep.complete;
                let mut pending = pending_sweep.items;

                if let Some(state) = state {
                    pending.retain(|order| order.state == state);
                }

                self.resolve_close_fraction_sizes(&mut pending)
                    .await
                    .map_err(OKXPendingAlgoOrderReportsError::new)?;

                let pending_reports_complete = match self.collect_algo_reports(
                    account_id,
                    &pending,
                    require_complete_active_coverage,
                    &mut instruments_cache,
                    ts_init,
                    start_ns,
                    end_ns,
                    &mut seen,
                    &mut reports,
                    &mut ambiguous_triggered_child_ids,
                ) {
                    Ok(complete) => complete,
                    Err(e) if require_complete_active_coverage => {
                        return Err(OKXPendingAlgoOrderReportsError::new(e).into());
                    }
                    Err(e) => return Err(e),
                };

                if require_complete_active_coverage && !pending_reports_complete {
                    return Err(OKXPendingAlgoOrderReportsError::new(anyhow::anyhow!(
                        "Pending {ord_type:?} algo order reports for {inst_type:?} could not be completely converted"
                    ))
                    .into());
                }
                complete &= pending_reports_complete;

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

            for history_state in history_states {
                params.state = Some(*history_state);
                let remaining = limit.map(|l| (l as usize).saturating_sub(reports.len()));

                let history_sweep = match self
                    .paginate_algo_history(
                        &params,
                        remaining,
                        require_complete_active_coverage,
                        reconciliation_history_page_limit(max_history_pages),
                    )
                    .await
                {
                    Ok(sweep) => sweep,
                    Err(e) if require_complete_active_coverage => {
                        log::warn!(
                            "Failed to fetch {history_state:?} {ord_type:?} algo order history for {inst_type:?}: {e}"
                        );
                        complete = false;
                        continue;
                    }
                    Err(e) => return Err(e),
                };
                complete &= history_sweep.complete;
                let mut history = history_sweep.items;

                if let Some(state) = state {
                    history.retain(|order| order.state == state);
                }

                complete &= self.collect_algo_reports(
                    account_id,
                    &history,
                    false,
                    &mut instruments_cache,
                    ts_init,
                    start_ns,
                    end_ns,
                    &mut seen,
                    &mut reports,
                    &mut ambiguous_triggered_child_ids,
                )?;

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

        Ok(AlgoOrderReportSweep {
            reports,
            complete,
            ambiguous_triggered_child_ids,
        })
    }

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

    /// Sets `sz` on open `closeFraction` algo orders from the positions they close.
    ///
    /// OKX omits `sz` when an algo order closes the whole position, and links the order to that
    /// position through `closeOrderAlgo` instead. Reconciliation needs a positive quantity to
    /// materialize the order, so the linked position size stands in for it.
    async fn resolve_close_fraction_sizes(
        &self,
        orders: &mut [OKXOrderAlgo],
    ) -> anyhow::Result<()> {
        let needs_size = |order: &OKXOrderAlgo| {
            order.sz.is_empty() && !order.close_fraction.is_empty() && is_open_okx_algo(order.state)
        };

        if !orders.iter().any(needs_size) {
            return Ok(());
        }

        let positions = self
            .inner
            .get_positions(GetPositionsParams::default())
            .await
            .map_err(|e| anyhow::anyhow!(e))?;

        for order in orders.iter_mut().filter(|order| needs_size(order)) {
            let position = positions.iter().find(|position| {
                position
                    .close_order_algo
                    .iter()
                    .any(|algo| algo.algo_id == order.algo_id)
            });

            let Some(position) = position else {
                log::warn!(
                    "No position links close-fraction algo order {}, quantity remains zero",
                    order.algo_id,
                );
                continue;
            };

            let size = Decimal::from_str(&position.pos).with_context(|| {
                format!("invalid position size for algo order {}", order.algo_id)
            })?;

            order.sz = size.abs().to_string();
        }

        Ok(())
    }

    #[expect(clippy::too_many_arguments)]
    fn collect_algo_reports(
        &self,
        account_id: AccountId,
        orders: &[OKXOrderAlgo],
        from_pending_source: bool,
        instruments_cache: &mut AHashMap<Ustr, InstrumentAny>,
        ts_init: UnixNanos,
        start_ns: Option<UnixNanos>,
        end_ns: Option<UnixNanos>,
        seen: &mut AHashMap<(String, String), usize>,
        reports: &mut Vec<OrderStatusReport>,
        ambiguous_triggered_child_ids: &mut AHashSet<VenueOrderId>,
    ) -> anyhow::Result<bool> {
        let mut complete = true;

        for order in orders {
            let key = (order.algo_id.clone(), order.algo_cl_ord_id.clone());

            let has_ambiguous_children = order.ord_id_list.len() > 1
                || !order.sub_algo_id_list.is_empty()
                || matches!(
                    order.ord_id_list.as_slice(),
                    [child_order_id]
                        if !order.ord_id.is_empty() && child_order_id != &order.ord_id
                );

            // Pending-source records are authoritative regardless of age or
            // mapped state; only terminal history respects the report window.
            if !from_pending_source
                && report_ts_outside_window(order.u_time, start_ns, end_ns)
                && !is_open_okx_algo(order.state)
            {
                continue;
            }

            let instrument = if let Some(instrument) = instruments_cache.get(&order.inst_id) {
                instrument.clone()
            } else {
                match self.resolve_report_instrument(
                    order.inst_id,
                    order.inst_type,
                    false,
                    is_open_okx_algo(order.state),
                    None,
                )? {
                    InstrumentResolution::Found(instrument) => {
                        instruments_cache.insert(order.inst_id, (*instrument).clone());
                        *instrument
                    }
                    InstrumentResolution::Skip => continue,
                    InstrumentResolution::Incomplete => {
                        complete = false;
                        continue;
                    }
                }
            };

            match parse_http_algo_order(order, account_id, &instrument, ts_init) {
                Ok(report) => {
                    if has_ambiguous_children && report.order_status == OrderStatus::Triggered {
                        log::warn!(
                            "Algo order {} has ambiguous child order identifiers",
                            order.algo_id,
                        );
                        ambiguous_triggered_child_ids.insert(report.venue_order_id);
                    }

                    if let Some(index) = seen.get(&key).copied() {
                        if is_order_status_report_more_advanced(&report, &reports[index]) {
                            reports[index] = report;
                        }
                    } else {
                        seen.insert(key, reports.len());
                        reports.push(report);
                    }
                }
                Err(e) => {
                    log::warn!("Failed to parse algo order report: {e}");
                    complete = false;
                }
            }
        }

        Ok(complete)
    }

    fn resolve_report_instrument(
        &self,
        symbol: Ustr,
        inst_type: OKXInstrumentType,
        is_spread: bool,
        is_open_or_position: bool,
        scope: Option<ReportInstrumentScope<'_>>,
    ) -> anyhow::Result<InstrumentResolution> {
        if let Ok(instrument) = self.instrument_from_cache(symbol) {
            return Ok(InstrumentResolution::Found(Box::new(instrument)));
        }

        let in_scope = match scope {
            None => true,
            Some(scope) if is_spread => scope.load_spreads,
            Some(scope) => {
                scope.instrument_types.contains(&inst_type)
                    || scope.instrument_types.contains(&OKXInstrumentType::Any)
            }
        };

        if !in_scope {
            log::debug!("Skipping report for out-of-scope instrument: symbol={symbol}");
            return Ok(InstrumentResolution::Skip);
        }

        if is_open_or_position {
            anyhow::bail!("Instrument {symbol} missing from cache");
        }

        log::warn!("Instrument {symbol} missing from cache");
        Ok(InstrumentResolution::Incomplete)
    }

    pub(crate) async fn request_spread_order_status_report(
        &self,
        account_id: AccountId,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> anyhow::Result<Option<OrderStatusReport>> {
        let instrument = self.instrument_from_cache(instrument_id.symbol.inner())?;
        let mut params_builder = GetSpreadOrderParamsBuilder::default();

        match (client_order_id, venue_order_id) {
            (Some(client_order_id), None) => {
                params_builder.cl_ord_id(client_order_id.as_str().to_string());
            }
            (None, Some(venue_order_id)) => {
                params_builder.ord_id(venue_order_id.as_str().to_string());
            }
            _ => anyhow::bail!(
                "Exactly one of client_order_id or venue_order_id is required for a spread order detail request"
            ),
        }

        let params = params_builder
            .build()
            .map_err(|e| anyhow::anyhow!("Failed to build spread order detail params: {e}"))?;
        let orders = match self.inner.get_spread_order(params).await {
            Ok(orders) => orders,
            Err(e) if e.is_order_not_found() => return Ok(None),
            Err(e) => return Err(e.into()),
        };
        let Some(order) = orders.into_iter().next() else {
            return Ok(None);
        };
        let ts_init = self.generate_ts_init();
        let report = parse_spread_order_status_report(
            &order,
            account_id,
            instrument.id(),
            instrument.price_precision(),
            instrument.size_precision(),
            ts_init,
        )?;

        Ok(Some(report))
    }
}

fn parse_historical_candlestick(
    raw: &OKXCandlestick,
    bar_type: BarType,
    price_precision: u8,
    size_precision: u8,
) -> anyhow::Result<Bar> {
    let mut bar = parse_candlestick(
        raw,
        bar_type,
        price_precision,
        size_precision,
        UnixNanos::default(),
    )?;

    bar.ts_init = if bar_type.spec().aggregation == BarAggregation::Month {
        let time_zone = jiff::tz::TimeZone::fixed(jiff::tz::offset(8));
        let (_, _, close) = get_time_bar_bounds(
            bar.ts_event.to_datetime_utc(),
            &bar_type,
            None,
            &time_zone,
            None,
        );
        UnixNanos::from(close)
    } else {
        bar.ts_event + get_bar_interval_ns(&bar_type)
    };

    Ok(bar)
}

fn is_open_okx_order(state: OKXOrderStatus) -> bool {
    matches!(
        state,
        OKXOrderStatus::Live | OKXOrderStatus::PartiallyFilled
    )
}

fn fill_quantity_is_positive(value: &str) -> Result<bool, rust_decimal::Error> {
    Ok(Decimal::from_str(value)? > Decimal::ZERO)
}

fn report_ts_outside_window(
    timestamp_ms: u64,
    start_ns: Option<UnixNanos>,
    end_ns: Option<UnixNanos>,
) -> bool {
    if start_ns.is_none() && end_ns.is_none() {
        return false;
    }

    let ts = parse_millisecond_timestamp(timestamp_ms);
    start_ns.is_some_and(|start| ts < start) || end_ns.is_some_and(|end| ts > end)
}

fn is_open_okx_algo(state: OKXAlgoOrderStatus) -> bool {
    matches!(
        state,
        OKXAlgoOrderStatus::Live
            | OKXAlgoOrderStatus::Pause
            | OKXAlgoOrderStatus::OrderPlaced
            | OKXAlgoOrderStatus::PartiallyEffective
    )
}

fn spread_page_limit(limit: Option<u32>) -> Option<u32> {
    limit.map(|limit| limit.min(OKX_PAGE_SIZE as u32))
}

fn parse_http_algo_order(
    order: &OKXOrderAlgo,
    account_id: AccountId,
    instrument: &InstrumentAny,
    ts_init: UnixNanos,
) -> anyhow::Result<OrderStatusReport> {
    let ord_id = if order.ord_id.is_empty() {
        match order.ord_id_list.as_slice() {
            [child_order_id] => child_order_id.clone(),
            _ => String::new(),
        }
    } else {
        order.ord_id.clone()
    };
    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,
        ord_id_list: order.ord_id_list.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(),
        fail_code: String::new(),
        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)
}

fn reconciliation_history_page_limit(max_pages: Option<u32>) -> usize {
    max_pages.map_or(MAX_RECONCILIATION_PAGES, |pages| {
        (pages as usize).min(MAX_RECONCILIATION_PAGES)
    })
}