nautilus_hyperliquid/http/

client.rs

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

//! Provides the HTTP client integration for the [Hyperliquid](https://hyperliquid.xyz/) REST API.
//!
//! This module defines and implements a [`HyperliquidHttpClient`] for sending requests to various
//! Hyperliquid endpoints. It handles request signing (when credentials are provided), constructs
//! valid HTTP requests using the [`HttpClient`], and parses the responses back into structured
//! data or an [`Error`].

use std::{
    collections::HashMap,
    num::NonZeroU32,
    sync::{Arc, LazyLock},
    time::Duration,
};

use ahash::{AHashMap, AHashSet};
use anyhow::Context;
use nautilus_common::cache::InstrumentLookupError;
use nautilus_core::{
    AtomicMap, UUID4, UnixNanos,
    datetime::datetime_to_unix_nanos,
    string::secret::SecretString,
    time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_model::{
    data::{Bar, BarType, FundingRateUpdate},
    enums::{
        AccountType, BarAggregation, CurrencyType, MarketStatusAction, OrderSide, OrderStatus,
        OrderType, TimeInForce, TriggerType,
    },
    events::AccountState,
    identifiers::{AccountId, ClientOrderId, InstrumentId, Symbol, VenueOrderId},
    instruments::{CurrencyPair, Instrument, InstrumentAny},
    orders::{Order, OrderAny},
    reports::{FillReport, OrderStatusReport, PositionStatusReport},
    types::{AccountBalance, Currency, Price, Quantity},
};
use nautilus_network::{
    http::{
        HttpClient, HttpClientError, HttpRedirectPolicy, HttpResponse, Method,
        create_standard_nautilus_headers, parse_retry_after,
    },
    ratelimiter::quota::Quota,
};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use serde_json::Value;
use ustr::Ustr;

use crate::{
    account::{normalize_account_address, resolve_execution_account_address},
    common::{
        consts::{
            ASSET_INDEX_INFO_KEY, HYPERLIQUID_REST_WEIGHT_PER_MINUTE, HYPERLIQUID_VENUE,
            NAUTILUS_BUILDER_ADDRESS, exchange_url, info_url,
        },
        credential::{Secrets, VaultAddress, credential_env_vars},
        enums::{
            HyperliquidAccountAbstraction, HyperliquidBarInterval, HyperliquidEnvironment,
            HyperliquidOrderStatus as HyperliquidOrderStatusEnum, HyperliquidProductType,
        },
        parse::{
            bar_type_to_interval, cache_alias_for_symbol, clamp_price_to_precision,
            derive_limit_from_trigger, determine_order_list_grouping, extract_inner_error,
            normalize_or_validate_wire_price, order_to_hyperliquid_request_with_optional_decimals,
            parse_combined_account_balances_and_margins, parse_outcome_symbol,
            parse_spot_account_balances, parse_trigger_order_type, parse_trigger_order_type_label,
            round_to_sig_figs, time_in_force_to_hyperliquid_tif,
        },
    },
    data::{candle_to_bar, funding_entry_to_update},
    data_types::HyperliquidPublicTrade,
    http::{
        error::{Error, Result},
        models::{
            ClearinghouseState, Cloid, HyperliquidCandleSnapshot, HyperliquidExchangeAction,
            HyperliquidExchangeBuilderFee, HyperliquidExchangeCancelByCloidRequest,
            HyperliquidExchangeCancelOrderRequest, HyperliquidExchangeGrouping,
            HyperliquidExchangeLimitParams, HyperliquidExchangeMergeOutcomeParams,
            HyperliquidExchangeMergeQuestionParams, HyperliquidExchangeModifyOrderRequest,
            HyperliquidExchangeModifyTarget, HyperliquidExchangeNegateOutcomeParams,
            HyperliquidExchangeOrderKind, HyperliquidExchangeOrderResponseData,
            HyperliquidExchangeOrderStatus, HyperliquidExchangePlaceOrderRequest,
            HyperliquidExchangeRequest, HyperliquidExchangeResponse,
            HyperliquidExchangeResponseData, HyperliquidExchangeSplitOutcomeParams,
            HyperliquidExchangeTif, HyperliquidExchangeTpSl, HyperliquidExchangeTriggerParams,
            HyperliquidExchangeUserOutcomeOp, HyperliquidFills, HyperliquidFundingHistoryEntry,
            HyperliquidL2Book, HyperliquidMeta, HyperliquidOrderStatus,
            HyperliquidOrderStatusEntry, HyperliquidRecentTrade, OutcomeMeta, PerpDex, PerpMeta,
            PerpMetaAndCtxs, RESPONSE_STATUS_OK, SpotClearinghouseState, SpotMeta, SpotMetaAndCtxs,
        },
        parse::{
            DEFAULT_OUTCOME_QUOTE_CURRENCY, HyperliquidInstrumentDef, create_instrument_from_def,
            filter_recent_public_trades, get_outcome_currency, instruments_from_defs_owned,
            parse_fill_report, parse_order_status_report_from_basic, parse_outcome_instruments,
            parse_perp_instruments_with_settlement, parse_position_status_report,
            parse_recent_public_trade, parse_spot_instruments, parse_spot_position_status_report,
            parse_unlisted_outcome_instrument, resolve_perp_settlement_currency,
        },
        query::{ExchangeAction, InfoRequest},
        rate_limits::{
            RateLimitSnapshot, WeightedLimiter, backoff_full_jitter, exchange_weight,
            exec_action_weight, info_base_weight, info_extra_weight, shared_rest_limiter,
        },
    },
    signing::{
        HyperliquidActionType, HyperliquidEip712Signer, NonceManager, SignRequest, types::SignerId,
    },
    websocket::messages::WsBasicOrderData,
};

fn deduplicate_historical_order_reports(reports: Vec<OrderStatusReport>) -> Vec<OrderStatusReport> {
    let mut best_by_venue_order_id = AHashMap::new();

    for candidate in reports {
        let Some(current) = best_by_venue_order_id.remove(&candidate.venue_order_id) else {
            best_by_venue_order_id.insert(candidate.venue_order_id, candidate);
            continue;
        };
        let (mut best, other) = if historical_report_is_more_advanced(&candidate, &current) {
            (candidate, current)
        } else {
            (current, candidate)
        };

        if matches!(
            best.order_type,
            OrderType::Limit | OrderType::StopLimit | OrderType::LimitIfTouched
        ) {
            best.price = best.price.or(other.price);
        }
        best.trigger_price = best.trigger_price.or(other.trigger_price);
        best.trigger_type = best.trigger_type.or(other.trigger_type);
        best_by_venue_order_id.insert(best.venue_order_id, best);
    }

    best_by_venue_order_id
        .into_values()
        .map(untrigger_closed_report_without_trigger_price)
        .collect()
}

// The venue zeroes `triggerPx` on the final row of a triggered order that has filled, so when no
// earlier row survives the order's trigger cannot be rebuilt and a conditional order type would
// fail initialization. A closed order no longer needs its trigger: report it as the limit order it
// executed as when a limit price survived, otherwise as a market order, so its fills reconcile.
fn untrigger_closed_report_without_trigger_price(
    mut report: OrderStatusReport,
) -> OrderStatusReport {
    let is_trigger_type = matches!(
        report.order_type,
        OrderType::StopMarket
            | OrderType::StopLimit
            | OrderType::MarketIfTouched
            | OrderType::LimitIfTouched
    );

    if !is_trigger_type || report.trigger_price.is_some() || !report.order_status.is_closed() {
        return report;
    }

    let order_type = match report.order_type {
        OrderType::StopLimit | OrderType::LimitIfTouched if report.price.is_some() => {
            OrderType::Limit
        }
        _ => {
            // A market order has no price; a leftover limit price would be used to infer fills
            report.price = None;
            OrderType::Market
        }
    };
    log::debug!(
        "Historical {} order {} has no trigger price, reconciling as {order_type}",
        report.order_type,
        report.venue_order_id,
    );
    report.order_type = order_type;
    report
}

fn historical_report_is_more_advanced(
    candidate: &OrderStatusReport,
    current: &OrderStatusReport,
) -> bool {
    candidate.filled_qty > current.filled_qty
        || (candidate.filled_qty == current.filled_qty
            && (historical_status_priority(candidate.order_status)
                > historical_status_priority(current.order_status)
                || (candidate.order_status == current.order_status
                    && candidate.ts_last > current.ts_last)))
}

const fn historical_status_priority(status: OrderStatus) -> u8 {
    match status {
        OrderStatus::Initialized | OrderStatus::Submitted | OrderStatus::Emulated => 0,
        OrderStatus::Released | OrderStatus::Denied => 1,
        OrderStatus::Accepted | OrderStatus::PendingUpdate | OrderStatus::PendingCancel => 2,
        OrderStatus::Triggered => 3,
        OrderStatus::PartiallyFilled => 4,
        OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Rejected => 5,
        OrderStatus::Filled | OrderStatus::Voided => 6,
    }
}

/// Unweighted REST quota retained for compatibility with existing callers.
///
/// Adapter clients use a shared weighted limiter because Hyperliquid aggregates request weights
/// across `/info` and `/exchange`.
pub static HYPERLIQUID_REST_QUOTA: LazyLock<Quota> = LazyLock::new(|| {
    Quota::per_minute(NonZeroU32::new(HYPERLIQUID_REST_WEIGHT_PER_MINUTE).unwrap())
});

pub(crate) const HYPERLIQUID_RECENT_HISTORY_LIMIT: usize = 2_000;
const RATE_LIMIT_BACKOFF_BASE: Duration = Duration::from_millis(125);
const RATE_LIMIT_BACKOFF_CAP: Duration = Duration::from_secs(5);
const RATE_LIMIT_INFO_RETRIES_MAX: u32 = 3;
const RETRY_AFTER_HEADER: &str = "retry-after";
const VAULT_TOKEN_PREFIX: &str = "vntls:";

/// Maximum rows returned per `fundingHistory` call (the venue per-response cap).
pub const HYPERLIQUID_FUNDING_HISTORY_MAX_LIMIT: usize = 500;
/// Seven days of hourly funding, used when a request gives no start or limit.
const DEFAULT_FUNDING_RATES_LIMIT: usize = 7 * 24;

/// Provides a raw HTTP client for low-level Hyperliquid REST API operations.
///
/// This client handles HTTP infrastructure, request signing, and raw API calls
/// that closely match Hyperliquid endpoint specifications.
#[derive(Debug, Clone)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.adapters.hyperliquid", from_py_object)
)]
pub struct HyperliquidRawHttpClient {
    client: HttpClient,
    environment: HyperliquidEnvironment,
    base_info: String,
    base_exchange: String,
    signer: Option<HyperliquidEip712Signer>,
    nonce_manager: Option<Arc<NonceManager>>,
    vault_address: Option<VaultAddress>,
    proxy_url: Option<SecretString>,
    info_limiter: Arc<WeightedLimiter>,
    exchange_limiter: Arc<WeightedLimiter>,
    info_max_retries: u32,
    info_retry_delay_initial: Duration,
    info_retry_delay_max: Duration,
}

impl HyperliquidRawHttpClient {
    /// Creates a new [`HyperliquidRawHttpClient`] for public endpoints only.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client cannot be created.
    pub fn new(
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> std::result::Result<Self, HttpClientError> {
        let base_info = info_url(environment).to_string();
        let base_exchange = exchange_url(environment).to_string();
        let info_limiter = shared_rest_limiter(environment, &base_info, proxy_url.as_deref());
        let exchange_limiter =
            shared_rest_limiter(environment, &base_exchange, proxy_url.as_deref());

        Ok(Self {
            client: Self::build_http_client(timeout_secs, proxy_url.clone())?,
            environment,
            base_info,
            base_exchange,
            signer: None,
            nonce_manager: None,
            vault_address: None,
            proxy_url: proxy_url.map(SecretString::from),
            info_limiter,
            exchange_limiter,
            info_max_retries: RATE_LIMIT_INFO_RETRIES_MAX,
            info_retry_delay_initial: RATE_LIMIT_BACKOFF_BASE,
            info_retry_delay_max: RATE_LIMIT_BACKOFF_CAP,
        })
    }

    /// Creates a new [`HyperliquidRawHttpClient`] configured with credentials
    /// for authenticated requests.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client cannot be created.
    pub fn with_credentials(
        secrets: &Secrets,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> std::result::Result<Self, HttpClientError> {
        let signer = HyperliquidEip712Signer::new(&secrets.private_key)
            .map_err(|e| HttpClientError::from(e.to_string()))?;
        let nonce_manager = Arc::new(NonceManager::new());
        let base_info = info_url(secrets.environment).to_string();
        let base_exchange = exchange_url(secrets.environment).to_string();
        let info_limiter =
            shared_rest_limiter(secrets.environment, &base_info, proxy_url.as_deref());
        let exchange_limiter =
            shared_rest_limiter(secrets.environment, &base_exchange, proxy_url.as_deref());

        Ok(Self {
            client: Self::build_http_client(timeout_secs, proxy_url.clone())?,
            environment: secrets.environment,
            base_info,
            base_exchange,
            signer: Some(signer),
            nonce_manager: Some(nonce_manager),
            vault_address: secrets.vault_address,
            proxy_url: proxy_url.map(SecretString::from),
            info_limiter,
            exchange_limiter,
            info_max_retries: RATE_LIMIT_INFO_RETRIES_MAX,
            info_retry_delay_initial: RATE_LIMIT_BACKOFF_BASE,
            info_retry_delay_max: RATE_LIMIT_BACKOFF_CAP,
        })
    }

    /// Overrides the base info URL (for testing with mock servers).
    pub fn set_base_info_url(&mut self, url: String) {
        self.info_limiter = shared_rest_limiter(
            self.environment,
            &url,
            self.proxy_url.as_ref().map(|value| value.expose_secret()),
        );
        self.base_info = url;
    }

    /// Overrides the base exchange URL (for testing with mock servers).
    pub fn set_base_exchange_url(&mut self, url: String) {
        self.exchange_limiter = shared_rest_limiter(
            self.environment,
            &url,
            self.proxy_url.as_ref().map(|value| value.expose_secret()),
        );
        self.base_exchange = url;
    }

    /// Creates an authenticated client from environment variables for the specified network.
    ///
    /// # Errors
    ///
    /// Returns an error if required environment variables are not set.
    pub fn from_env(environment: HyperliquidEnvironment) -> Result<Self> {
        let secrets = Secrets::from_env(environment)
            .map_err(|e| Error::auth(format!("missing credentials in environment: {e}")))?;
        Self::with_credentials(&secrets, 60, None)
            .map_err(|e| Error::auth(format!("Failed to create HTTP client: {e}")))
    }

    /// Creates a new [`HyperliquidRawHttpClient`] configured with explicit credentials.
    ///
    /// # Errors
    ///
    /// Returns an error if the private key is invalid or cannot be parsed.
    pub fn from_credentials(
        private_key: &str,
        vault_address: Option<&str>,
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> Result<Self> {
        let secrets = Secrets::from_private_key(private_key, vault_address, environment)
            .map_err(|e| Error::auth(format!("invalid credentials: {e}")))?;
        Self::with_credentials(&secrets, timeout_secs, proxy_url)
            .map_err(|e| Error::auth(format!("Failed to create HTTP client: {e}")))
    }

    /// Rebinds the client to the shared rate limits for its configured routes.
    #[must_use]
    pub fn with_rate_limits(mut self) -> Self {
        let proxy_url = self.proxy_url.as_ref().map(|value| value.expose_secret());
        self.info_limiter = shared_rest_limiter(self.environment, &self.base_info, proxy_url);
        self.exchange_limiter =
            shared_rest_limiter(self.environment, &self.base_exchange, proxy_url);
        self
    }

    /// Returns the configured environment.
    #[must_use]
    pub fn environment(&self) -> HyperliquidEnvironment {
        self.environment
    }

    /// Returns whether this client is configured for testnet.
    #[must_use]
    pub fn is_testnet(&self) -> bool {
        self.environment == HyperliquidEnvironment::Testnet
    }

    /// Gets the user address derived from the private key (if client has credentials).
    ///
    /// # Errors
    ///
    /// Returns an error if the client has no signer configured.
    pub fn get_user_address(&self) -> Result<String> {
        self.signer
            .as_ref()
            .ok_or_else(|| Error::auth("No signer configured"))?
            .address()
    }

    /// Returns `true` if a vault address is configured.
    #[must_use]
    pub fn has_vault_address(&self) -> bool {
        self.vault_address.is_some()
    }

    /// Gets the account address for queries: vault address if configured,
    /// otherwise the user (EOA) address.
    ///
    /// # Errors
    ///
    /// Returns an error if the client has no signer configured.
    pub fn get_account_address(&self) -> Result<String> {
        if let Some(vault) = &self.vault_address {
            Ok(vault.to_hex())
        } else {
            self.get_user_address()
        }
    }

    fn build_http_client(
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> std::result::Result<HttpClient, HttpClientError> {
        HttpClient::builder()
            .redirect_policy(HttpRedirectPolicy::Reject)
            .headers(Self::default_headers())
            .header_keys(vec![RETRY_AFTER_HEADER.to_string()])
            .rate_limiters(Vec::new())
            .timeout_secs(timeout_secs)
            .maybe_proxy_url(proxy_url)
            .build()
    }

    fn default_headers() -> HashMap<String, String> {
        let mut headers: HashMap<String, String> =
            create_standard_nautilus_headers().into_iter().collect();
        headers.insert("Content-Type".to_string(), "application/json".to_string());
        headers
    }

    fn signer_id(&self) -> SignerId {
        SignerId("hyperliquid:default".into())
    }

    fn retry_after_ms(headers: &HashMap<String, String>) -> Option<u64> {
        let retry_after = headers.get(RETRY_AFTER_HEADER)?;
        parse_retry_after(retry_after, jiff::Timestamp::now())
            .map(|delay| delay.as_millis().min(u128::from(u64::MAX)) as u64)
    }

    /// Get metadata about available markets.
    pub async fn info_meta(&self) -> Result<HyperliquidMeta> {
        let request = InfoRequest::meta();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get complete spot metadata (tokens and pairs).
    pub async fn get_spot_meta(&self) -> Result<SpotMeta> {
        let request = InfoRequest::spot_meta();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get perpetuals metadata with asset contexts (for price precision refinement).
    pub async fn get_perp_meta_and_ctxs(&self) -> Result<PerpMetaAndCtxs> {
        let request = InfoRequest::meta_and_asset_ctxs();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get spot metadata with asset contexts (for price precision refinement).
    pub async fn get_spot_meta_and_ctxs(&self) -> Result<SpotMetaAndCtxs> {
        let request = InfoRequest::spot_meta_and_asset_ctxs();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get outcome metadata.
    pub async fn get_outcome_meta(&self) -> Result<OutcomeMeta> {
        let request = InfoRequest::outcome_meta();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    pub(crate) async fn load_perp_meta(&self) -> Result<PerpMeta> {
        let request = InfoRequest::meta();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get metadata for all perp dexes (standard + HIP-3).
    pub(crate) async fn load_all_perp_metas(&self) -> Result<Vec<PerpMeta>> {
        let request = InfoRequest::all_perp_metas();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get the list of perp dex names aligned by dex index.
    pub(crate) async fn load_perp_dexs(&self) -> Result<Vec<Option<PerpDex>>> {
        let request = InfoRequest::perp_dexs();
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get L2 order book for a coin.
    pub async fn info_l2_book(&self, coin: &str) -> Result<HyperliquidL2Book> {
        let request = InfoRequest::l2_book(coin);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get recent public trades for a coin.
    ///
    /// Returns a recent snapshot (newest first) with no time range. Depends on the
    /// Hyperliquid indexer: self-hosted `/info` nodes return HTTP 422.
    pub async fn info_recent_trades(&self, coin: &str) -> Result<Vec<HyperliquidRecentTrade>> {
        let request = InfoRequest::recent_trades(coin);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get user fills (trading history).
    pub async fn info_user_fills(&self, user: &str) -> Result<HyperliquidFills> {
        let request = InfoRequest::user_fills(user);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get order status for a user.
    pub async fn info_order_status(&self, user: &str, oid: u64) -> Result<HyperliquidOrderStatus> {
        let request = InfoRequest::order_status(user, oid);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get all open orders for a user.
    pub async fn info_open_orders(&self, user: &str) -> Result<Value> {
        let request = InfoRequest::open_orders(user);
        self.send_info_request(&request).await
    }

    /// Get frontend open orders (includes more detail) for a user.
    pub async fn info_frontend_open_orders(&self, user: &str) -> Result<Value> {
        self.info_frontend_open_orders_for_dex(user, None).await
    }

    async fn info_frontend_open_orders_for_dex(
        &self,
        user: &str,
        dex: Option<&str>,
    ) -> Result<Value> {
        let request = InfoRequest::frontend_open_orders_for_dex(user, dex);
        self.send_info_request(&request).await
    }

    /// Get the most recent historical orders for a user.
    pub async fn info_historical_orders(
        &self,
        user: &str,
    ) -> Result<Vec<HyperliquidOrderStatusEntry>> {
        let request = InfoRequest::historical_orders(user);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get clearinghouse state (balances, positions, margin) for a user.
    pub async fn info_clearinghouse_state(&self, user: &str) -> Result<Value> {
        self.info_clearinghouse_state_for_dex(user, None).await
    }

    async fn info_clearinghouse_state_for_dex(
        &self,
        user: &str,
        dex: Option<&str>,
    ) -> Result<Value> {
        let request = InfoRequest::clearinghouse_state_for_dex(user, dex);
        self.send_info_request(&request).await
    }

    /// Get spot clearinghouse state (per-token spot balances) for a user.
    pub async fn info_spot_clearinghouse_state(&self, user: &str) -> Result<Value> {
        let request = InfoRequest::spot_clearinghouse_state(user);
        self.send_info_request(&request).await
    }

    /// Get the account abstraction mode for a user.
    pub async fn info_user_abstraction(&self, user: &str) -> Result<Value> {
        let request = InfoRequest::user_abstraction(user);
        self.send_info_request(&request).await
    }

    /// Get user fee schedule and effective rates.
    pub async fn info_user_fees(&self, user: &str) -> Result<Value> {
        let request = InfoRequest::user_fees(user);
        self.send_info_request(&request).await
    }

    /// Get candle/bar data for a coin.
    pub async fn info_candle_snapshot(
        &self,
        coin: &str,
        interval: HyperliquidBarInterval,
        start_time: u64,
        end_time: u64,
    ) -> Result<HyperliquidCandleSnapshot> {
        let request = InfoRequest::candle_snapshot(coin, interval, start_time, end_time);
        let response = self.send_info_request(&request).await?;

        log::trace!(
            "Candle snapshot raw response (len={}): {:?}",
            response.as_array().map_or(0, |a| a.len()),
            response
        );

        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Get historical funding rates for a coin.
    ///
    /// `start_time` and `end_time` are Unix milliseconds. `end_time` is optional;
    /// if omitted, the venue returns entries up to the most recent funding.
    pub async fn info_funding_history(
        &self,
        coin: &str,
        start_time: u64,
        end_time: Option<u64>,
    ) -> Result<Vec<HyperliquidFundingHistoryEntry>> {
        let request = InfoRequest::funding_history(coin, start_time, end_time);
        let response = self.send_info_request(&request).await?;
        serde_json::from_value(response).map_err(Error::Serde)
    }

    /// Generic info request method that returns raw JSON (useful for new endpoints and testing).
    pub async fn send_info_request_raw(&self, request: &InfoRequest) -> Result<Value> {
        self.send_info_request(request).await
    }

    async fn send_info_request(&self, request: &InfoRequest) -> Result<Value> {
        let base_w = info_base_weight(request);
        let mut attempt = 0u32;

        loop {
            self.info_limiter.acquire(base_w).await;

            let (e, retry_after_ms) = match self.http_roundtrip_info(request).await {
                Ok(response) if response.status.is_success() => {
                    let val: Value =
                        serde_json::from_slice(&response.body).map_err(Error::Serde)?;
                    let extra = info_extra_weight(request, &val);
                    if extra > 0 {
                        self.info_limiter.debit_extra(extra).await;
                        log::debug!(
                            "Info debited extra weight: endpoint={request:?}, base_w={base_w}, extra={extra}"
                        );
                    }

                    return Ok(val);
                }
                Ok(response) => {
                    let retry_after_ms = Self::retry_after_ms(&response.headers);
                    let error = if response.status.as_u16() == 429 {
                        Error::rate_limit("info", base_w, retry_after_ms)
                    } else {
                        let body = String::from_utf8_lossy(&response.body);
                        Error::http(response.status.as_u16(), body.to_string())
                    };

                    (error, retry_after_ms)
                }
                Err(e) => (e, None),
            };

            if !e.is_retryable() || attempt >= self.info_max_retries {
                return Err(e);
            }

            let delay = retry_after_ms.map_or_else(
                || {
                    backoff_full_jitter(
                        attempt,
                        self.info_retry_delay_initial,
                        self.info_retry_delay_max,
                    )
                },
                Duration::from_millis,
            );

            log::debug!(
                "Info request failed; retrying: endpoint={request:?}, attempt={attempt}, wait_ms={}",
                delay.as_millis()
            );
            attempt += 1;
            tokio::time::sleep(delay).await;
        }
    }

    async fn http_roundtrip_info(&self, request: &InfoRequest) -> Result<HttpResponse> {
        let url = &self.base_info;
        let body = serde_json::to_value(request).map_err(Error::Serde)?;
        let body_bytes = serde_json::to_string(&body)
            .map_err(Error::Serde)?
            .into_bytes();

        self.client
            .request(
                Method::POST,
                url.clone(),
                None,
                None,
                Some(body_bytes),
                None,
                None,
            )
            .await
            .map_err(Error::from_http_client)
    }

    /// Send a signed action to the exchange.
    pub async fn post_action(
        &self,
        action: &ExchangeAction,
    ) -> Result<HyperliquidExchangeResponse> {
        let w = exchange_weight(action);
        self.exchange_limiter.acquire(w).await;

        let signer = self
            .signer
            .as_ref()
            .ok_or_else(|| Error::auth("credentials required for exchange operations"))?;

        let nonce_manager = self
            .nonce_manager
            .as_ref()
            .ok_or_else(|| Error::auth("nonce manager missing"))?;

        let signer_id = self.signer_id();
        let time_nonce = nonce_manager.next(signer_id)?;

        // L1 signing uses `action_bytes` only; skip the JSON value to save work
        let action_bytes = rmp_serde::to_vec_named(action)
            .context("serialize action with MessagePack")
            .map_err(|e| Error::bad_request(e.to_string()))?;

        let sign_request = SignRequest {
            action: None,
            action_bytes: Some(action_bytes),
            time_nonce,
            action_type: HyperliquidActionType::L1,
            is_testnet: self.is_testnet(),
            vault_address: self.vault_address,
            expires_after: None,
        };

        let sig = signer.sign(&sign_request)?.signature;

        let nonce_u64 = time_nonce.as_millis() as u64;

        let request = if let Some(vault) = self.vault_address {
            HyperliquidExchangeRequest::with_vault(
                action.clone(),
                nonce_u64,
                sig,
                vault.to_string(),
            )
        } else {
            HyperliquidExchangeRequest::new(action.clone(), nonce_u64, sig)
        };

        let response = self.http_roundtrip_exchange(&request).await?;

        if response.status.is_success() {
            let parsed_response: HyperliquidExchangeResponse =
                serde_json::from_slice(&response.body).map_err(Error::Serde)?;

            // Check if the response contains an error status
            match &parsed_response {
                HyperliquidExchangeResponse::Status {
                    status,
                    response: response_data,
                } if status == "err" => {
                    let error_msg = response_data
                        .as_str()
                        .map_or_else(|| response_data.to_string(), |s| s.to_string());
                    log::error!("Hyperliquid API returned error: {error_msg}");
                    Err(Error::bad_request(format!("API error: {error_msg}")))
                }
                HyperliquidExchangeResponse::Error { error } => {
                    log::error!("Hyperliquid API returned error: {error}");
                    Err(Error::bad_request(format!("API error: {error}")))
                }
                _ => Ok(parsed_response),
            }
        } else if response.status.as_u16() == 429 {
            let ra = Self::retry_after_ms(&response.headers);
            Err(Error::rate_limit("exchange", w, ra))
        } else {
            let error_body = String::from_utf8_lossy(&response.body);
            log::error!(
                "Exchange API error (status {}): {}",
                response.status.as_u16(),
                error_body
            );
            Err(Error::http(
                response.status.as_u16(),
                error_body.to_string(),
            ))
        }
    }

    /// Build a signed exchange request using the typed HyperliquidExchangeAction enum.
    ///
    /// # Errors
    ///
    /// Returns an error if the action is unsupported, credentials are missing, or signing fails.
    pub fn sign_action_exec_request(
        &self,
        action: &HyperliquidExchangeAction,
        expires_after: Option<u64>,
    ) -> Result<HyperliquidExchangeRequest<HyperliquidExchangeAction>> {
        if matches!(action, HyperliquidExchangeAction::UsdClassTransfer { .. }) {
            return Err(Error::bad_request("Unsupported action: usdClassTransfer"));
        }

        let signer = self
            .signer
            .as_ref()
            .ok_or_else(|| Error::auth("credentials required for exchange operations"))?;

        let nonce_manager = self
            .nonce_manager
            .as_ref()
            .ok_or_else(|| Error::auth("nonce manager missing"))?;

        let signer_id = self.signer_id();
        let time_nonce = nonce_manager.next(signer_id)?;
        // No need to validate - next() guarantees a valid, unused nonce

        // L1 signing uses `action_bytes` only; skip the JSON value to save work
        let action_bytes = rmp_serde::to_vec_named(action)
            .context("serialize action with MessagePack")
            .map_err(|e| Error::bad_request(e.to_string()))?;

        let sig = signer
            .sign(&SignRequest {
                action: None,
                action_bytes: Some(action_bytes),
                time_nonce,
                action_type: HyperliquidActionType::L1,
                is_testnet: self.is_testnet(),
                vault_address: self.vault_address,
                expires_after,
            })?
            .signature;

        let mut request = if let Some(vault) = self.vault_address {
            HyperliquidExchangeRequest::with_vault(
                action.clone(),
                time_nonce.as_millis() as u64,
                sig,
                vault.to_string(),
            )
        } else {
            HyperliquidExchangeRequest::new(action.clone(), time_nonce.as_millis() as u64, sig)
        };
        request.expires_after = expires_after;
        Ok(request)
    }

    /// Send a signed action to the exchange using the typed HyperliquidExchangeAction enum.
    ///
    /// This is the preferred method for placing orders as it uses properly typed
    /// structures that match Hyperliquid's API expectations exactly.
    pub async fn post_action_exec(
        &self,
        action: &HyperliquidExchangeAction,
    ) -> Result<HyperliquidExchangeResponse> {
        let w = exec_action_weight(action);
        self.exchange_limiter.acquire(w).await;

        let request = self.sign_action_exec_request(action, None)?;

        let response = self.http_roundtrip_exchange(&request).await?;

        if response.status.is_success() {
            let parsed_response: HyperliquidExchangeResponse =
                serde_json::from_slice(&response.body).map_err(Error::Serde)?;

            // Check if the response contains an error status
            match &parsed_response {
                HyperliquidExchangeResponse::Status {
                    status,
                    response: response_data,
                } if status == "err" => {
                    let error_msg = response_data
                        .as_str()
                        .map_or_else(|| response_data.to_string(), |s| s.to_string());
                    log::error!("Hyperliquid API returned error: {error_msg}");
                    Err(Error::bad_request(format!("API error: {error_msg}")))
                }
                HyperliquidExchangeResponse::Error { error } => {
                    log::error!("Hyperliquid API returned error: {error}");
                    Err(Error::bad_request(format!("API error: {error}")))
                }
                _ => Ok(parsed_response),
            }
        } else if response.status.as_u16() == 429 {
            let ra = Self::retry_after_ms(&response.headers);
            Err(Error::rate_limit("exchange", w, ra))
        } else {
            let error_body = String::from_utf8_lossy(&response.body);
            Err(Error::http(
                response.status.as_u16(),
                error_body.to_string(),
            ))
        }
    }

    /// Returns the current rate-limit state for the info endpoint route.
    pub async fn rest_limiter_snapshot(&self) -> RateLimitSnapshot {
        self.info_limiter.snapshot().await
    }

    async fn http_roundtrip_exchange<T>(
        &self,
        request: &HyperliquidExchangeRequest<T>,
    ) -> Result<HttpResponse>
    where
        T: serde::Serialize,
    {
        let url = &self.base_exchange;
        let body = serde_json::to_string(&request).map_err(Error::Serde)?;
        let body_bytes = body.into_bytes();

        let response = self
            .client
            .request(
                Method::POST,
                url.clone(),
                None,
                None,
                Some(body_bytes),
                None,
                None,
            )
            .await
            .map_err(Error::from_http_client)?;

        Ok(response)
    }
}

/// Provides a high-level HTTP client for the [Hyperliquid](https://hyperliquid.xyz/) REST API.
///
/// This domain client wraps [`HyperliquidRawHttpClient`] and provides methods that work
/// with Nautilus domain types. It maintains an instrument cache and handles conversions
/// between Hyperliquid API responses and Nautilus domain models.
#[derive(Debug, Clone)]
#[cfg_attr(
    feature = "python",
    pyo3::pyclass(module = "nautilus_trader.adapters.hyperliquid", from_py_object)
)]
#[cfg_attr(
    feature = "python",
    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.hyperliquid")
)]
pub struct HyperliquidHttpClient {
    pub(crate) inner: Arc<HyperliquidRawHttpClient>,
    clock: &'static AtomicTime,
    instruments: Arc<AtomicMap<Ustr, InstrumentAny>>,
    instruments_by_coin: Arc<AtomicMap<(Ustr, HyperliquidProductType), InstrumentAny>>,
    /// Mapping from symbol to asset index for order submission.
    asset_indices: Arc<AtomicMap<Ustr, u32>>,
    /// Mapping from spot fill coin (`@{pair_index}`) to instrument symbol.
    spot_fill_coins: Arc<AtomicMap<Ustr, Ustr>>,
    client_order_id_cloids: Arc<Mutex<AHashMap<ClientOrderId, Cloid>>>,
    account_id: Option<AccountId>,
    /// Optional override address for queries (agent wallet / API sub-key support).
    /// When set, used for balance queries, position reports, and WS subscriptions
    /// instead of the address derived from the private key.
    account_address: Option<String>,
    normalize_prices: bool,
    market_order_slippage_bps: u32,
    include_builder_attribution: bool,
}

impl Default for HyperliquidHttpClient {
    fn default() -> Self {
        Self::new(HyperliquidEnvironment::Mainnet, 60, None)
            .expect("Failed to create default Hyperliquid HTTP client")
    }
}

impl HyperliquidHttpClient {
    /// Creates a new [`HyperliquidHttpClient`] for public endpoints only.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client cannot be created.
    pub fn new(
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> std::result::Result<Self, HttpClientError> {
        let raw_client = HyperliquidRawHttpClient::new(environment, timeout_secs, proxy_url)?;
        Ok(Self::from_raw(raw_client))
    }

    /// Creates a new [`HyperliquidHttpClient`] configured with a [`Secrets`] struct.
    ///
    /// # Errors
    ///
    /// Returns an error if the HTTP client cannot be created.
    pub fn with_secrets(
        secrets: &Secrets,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> std::result::Result<Self, HttpClientError> {
        let raw_client =
            HyperliquidRawHttpClient::with_credentials(secrets, timeout_secs, proxy_url)?;
        Ok(Self::from_raw(raw_client))
    }

    fn from_raw(raw_client: HyperliquidRawHttpClient) -> Self {
        Self {
            inner: Arc::new(raw_client),
            clock: get_atomic_clock_realtime(),
            instruments: Arc::new(AtomicMap::new()),
            instruments_by_coin: Arc::new(AtomicMap::new()),
            asset_indices: Arc::new(AtomicMap::new()),
            spot_fill_coins: Arc::new(AtomicMap::new()),
            client_order_id_cloids: Arc::new(Mutex::new(AHashMap::new())),
            account_id: None,
            account_address: None,
            normalize_prices: true,
            market_order_slippage_bps: crate::common::parse::DEFAULT_MARKET_SLIPPAGE_BPS,
            include_builder_attribution: true,
        }
    }

    /// Returns the cached CLOID for a client order ID, or derives and caches it.
    #[must_use]
    pub fn get_or_generate_client_order_id_cloid(&self, client_order_id: ClientOrderId) -> Cloid {
        let mut cloids = self.client_order_id_cloids.lock();
        *cloids
            .entry(client_order_id)
            .or_insert_with(|| Cloid::from_client_order_id(client_order_id))
    }

    /// Caches a CLOID for a client order ID if one is not already cached.
    pub fn cache_client_order_id_cloid(&self, client_order_id: ClientOrderId, cloid: Cloid) {
        self.client_order_id_cloids
            .lock()
            .entry(client_order_id)
            .or_insert(cloid);
    }

    /// Returns the cached CLOID for a client order ID.
    #[must_use]
    pub fn cached_client_order_id_cloid(&self, client_order_id: &ClientOrderId) -> Option<Cloid> {
        self.client_order_id_cloids
            .lock()
            .get(client_order_id)
            .copied()
    }

    /// Returns the cached CLOID for a client order ID when no other client
    /// order ID maps to the same CLOID.
    #[must_use]
    pub(crate) fn unique_cached_client_order_id_cloid(
        &self,
        client_order_id: &ClientOrderId,
    ) -> Option<Cloid> {
        let cloids = self.client_order_id_cloids.lock();
        let cloid = cloids.get(client_order_id).copied()?;
        let mapping_count = cloids
            .values()
            .filter(|cached_cloid| **cached_cloid == cloid)
            .count();

        (mapping_count == 1).then_some(cloid)
    }

    /// Removes the cached CLOID for a client order ID.
    pub fn remove_client_order_id_cloid(&self, client_order_id: &ClientOrderId) -> Option<Cloid> {
        self.client_order_id_cloids.lock().remove(client_order_id)
    }

    /// Overrides the base info URL (for testing with mock servers).
    ///
    /// # Panics
    ///
    /// Panics if the inner `Arc` has multiple references.
    pub fn set_base_info_url(&mut self, url: String) {
        Arc::get_mut(&mut self.inner)
            .expect("cannot override URL: Arc has multiple references")
            .set_base_info_url(url);
    }

    /// Overrides the base exchange URL (for testing with mock servers).
    ///
    /// # Panics
    ///
    /// Panics if the inner `Arc` has multiple references.
    pub fn set_base_exchange_url(&mut self, url: String) {
        Arc::get_mut(&mut self.inner)
            .expect("cannot override URL: Arc has multiple references")
            .set_base_exchange_url(url);
    }

    pub(crate) fn set_info_retry_config(
        &mut self,
        max_retries: u32,
        delay_initial: Duration,
        delay_max: Duration,
    ) {
        let inner = Arc::make_mut(&mut self.inner);
        inner.info_max_retries = max_retries;
        inner.info_retry_delay_initial = delay_initial;
        inner.info_retry_delay_max = delay_max;
    }

    /// Creates an authenticated client from environment variables for the specified network.
    ///
    /// # Errors
    ///
    /// Returns an error if required environment variables are not set.
    pub fn from_env(environment: HyperliquidEnvironment) -> Result<Self> {
        let raw_client = HyperliquidRawHttpClient::from_env(environment)?;
        Ok(Self {
            inner: Arc::new(raw_client),
            clock: get_atomic_clock_realtime(),
            instruments: Arc::new(AtomicMap::new()),
            instruments_by_coin: Arc::new(AtomicMap::new()),
            asset_indices: Arc::new(AtomicMap::new()),
            spot_fill_coins: Arc::new(AtomicMap::new()),
            client_order_id_cloids: Arc::new(Mutex::new(AHashMap::new())),
            account_id: None,
            account_address: None,
            normalize_prices: true,
            market_order_slippage_bps: crate::common::parse::DEFAULT_MARKET_SLIPPAGE_BPS,
            include_builder_attribution: true,
        })
    }

    /// Creates a new [`HyperliquidHttpClient`] configured with credentials.
    ///
    /// If credentials are not provided, falls back to environment variables:
    /// - Testnet: `HYPERLIQUID_TESTNET_PK`, `HYPERLIQUID_TESTNET_VAULT`
    /// - Mainnet: `HYPERLIQUID_PK`, `HYPERLIQUID_VAULT`
    ///
    /// If no credentials are provided and no environment variables are set,
    /// creates an unauthenticated client for public endpoints only.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are invalid.
    pub fn with_credentials(
        private_key: Option<String>,
        vault_address: Option<String>,
        account_address: Option<&str>,
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> Result<Self> {
        let (pk_env_var, vault_env_var) = credential_env_vars(environment);

        let resolved_account_address = resolve_execution_account_address(
            private_key.as_deref(),
            vault_address.as_deref(),
            account_address,
            environment,
        )?;

        // Resolve private key: explicit value -> env var -> None (unauthenticated)
        let resolved_pk = private_key.or_else(|| std::env::var(pk_env_var).ok());

        // Resolve vault address: explicit value -> env var -> None
        let resolved_vault = vault_address.or_else(|| std::env::var(vault_env_var).ok());

        Self::from_resolved_credentials(
            resolved_pk,
            resolved_vault.as_deref(),
            resolved_account_address,
            environment,
            timeout_secs,
            proxy_url,
        )
    }

    fn from_resolved_credentials(
        private_key: Option<String>,
        vault_address: Option<&str>,
        account_address: Option<String>,
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> Result<Self> {
        match private_key {
            Some(pk) => {
                let raw_client = HyperliquidRawHttpClient::from_credentials(
                    &pk,
                    vault_address,
                    environment,
                    timeout_secs,
                    proxy_url,
                )?;
                Ok(Self {
                    inner: Arc::new(raw_client),
                    clock: get_atomic_clock_realtime(),
                    instruments: Arc::new(AtomicMap::new()),
                    instruments_by_coin: Arc::new(AtomicMap::new()),
                    asset_indices: Arc::new(AtomicMap::new()),
                    spot_fill_coins: Arc::new(AtomicMap::new()),
                    client_order_id_cloids: Arc::new(Mutex::new(AHashMap::new())),
                    account_id: None,
                    account_address,
                    normalize_prices: true,
                    market_order_slippage_bps: crate::common::parse::DEFAULT_MARKET_SLIPPAGE_BPS,
                    include_builder_attribution: true,
                })
            }
            None => {
                // No credentials available, create unauthenticated client
                let mut client = Self::new(environment, timeout_secs, proxy_url)
                    .map_err(|e| Error::auth(format!("Failed to create HTTP client: {e}")))?;
                client.set_account_address(account_address)?;
                Ok(client)
            }
        }
    }

    /// Creates a new [`HyperliquidHttpClient`] configured with explicit credentials.
    ///
    /// # Errors
    ///
    /// Returns an error if the private key is invalid or cannot be parsed.
    pub fn from_credentials(
        private_key: &str,
        vault_address: Option<&str>,
        environment: HyperliquidEnvironment,
        timeout_secs: u64,
        proxy_url: Option<String>,
    ) -> Result<Self> {
        let raw_client = HyperliquidRawHttpClient::from_credentials(
            private_key,
            vault_address,
            environment,
            timeout_secs,
            proxy_url,
        )?;
        Ok(Self {
            inner: Arc::new(raw_client),
            clock: get_atomic_clock_realtime(),
            instruments: Arc::new(AtomicMap::new()),
            instruments_by_coin: Arc::new(AtomicMap::new()),
            asset_indices: Arc::new(AtomicMap::new()),
            spot_fill_coins: Arc::new(AtomicMap::new()),
            client_order_id_cloids: Arc::new(Mutex::new(AHashMap::new())),
            account_id: None,
            account_address: None,
            normalize_prices: true,
            market_order_slippage_bps: crate::common::parse::DEFAULT_MARKET_SLIPPAGE_BPS,
            include_builder_attribution: true,
        })
    }

    /// Returns whether this client is configured for testnet.
    #[must_use]
    pub fn is_testnet(&self) -> bool {
        self.inner.is_testnet()
    }

    /// Returns whether order price normalization is enabled.
    #[must_use]
    pub fn normalize_prices(&self) -> bool {
        self.normalize_prices
    }

    /// Sets whether to normalize order prices to 5 significant figures.
    pub fn set_normalize_prices(&mut self, value: bool) {
        self.normalize_prices = value;
    }

    /// Returns the MARKET-order slippage buffer in basis points.
    #[must_use]
    pub fn market_order_slippage_bps(&self) -> u32 {
        self.market_order_slippage_bps
    }

    /// Sets the MARKET-order slippage buffer in basis points.
    pub fn set_market_order_slippage_bps(&mut self, value: u32) {
        self.market_order_slippage_bps = value;
    }

    /// Returns whether eligible mainnet orders include builder attribution.
    #[must_use]
    pub fn include_builder_attribution(&self) -> bool {
        self.include_builder_attribution
    }

    /// Sets whether eligible mainnet orders include builder attribution.
    pub fn set_include_builder_attribution(&mut self, value: bool) {
        self.include_builder_attribution = value;
    }

    /// Gets the user address derived from the private key (if client has credentials).
    ///
    /// # Errors
    ///
    /// Returns an error if the client has no signer configured.
    pub fn get_user_address(&self) -> Result<String> {
        self.inner.get_user_address()
    }

    /// Returns `true` if a vault address is configured.
    #[must_use]
    pub fn has_vault_address(&self) -> bool {
        self.inner.has_vault_address()
    }

    /// Returns the builder-attribution fee to attach to outgoing orders.
    ///
    /// Returns `None` when attribution is disabled, or when Hyperliquid does
    /// not support it for the current request context (vault orders and testnet).
    #[must_use]
    pub fn builder_attribution(&self) -> Option<HyperliquidExchangeBuilderFee> {
        if !self.include_builder_attribution || self.has_vault_address() || self.is_testnet() {
            None
        } else {
            Some(HyperliquidExchangeBuilderFee {
                address: NAUTILUS_BUILDER_ADDRESS.to_string(),
                fee_tenths_bp: 0,
            })
        }
    }

    /// Gets the account address for queries: account_address if configured
    /// (agent wallet), then vault address, otherwise the user (EOA) address.
    ///
    /// # Errors
    ///
    /// Returns an error if the client has no signer configured and no account_address override is
    /// set.
    pub fn get_account_address(&self) -> Result<String> {
        if let Some(addr) = &self.account_address {
            return Ok(addr.clone());
        }
        self.inner.get_account_address()
    }

    /// Sets the account address override for queries (agent wallet support).
    ///
    /// Trims surrounding whitespace and stores lowercase hex. Requires a `0x`-prefixed lowercase
    /// or EIP-55 checksummed address.
    /// `None` clears the override. Invalid input leaves the current override unchanged.
    ///
    /// # Errors
    ///
    /// Returns an error if the account address is invalid.
    pub fn set_account_address(&mut self, address: Option<String>) -> Result<()> {
        self.account_address = address
            .map(|address| normalize_account_address(&address))
            .transpose()?;
        Ok(())
    }

    /// Caches a single instrument.
    ///
    /// This is required for parsing orders, fills, and positions into reports.
    /// Any existing instrument with the same symbol will be replaced.
    ///
    /// The venue asset index is taken from the instrument's `info` map so an
    /// instrument arriving on the message bus becomes submittable without
    /// refetching venue metadata. An instrument without the key keeps its
    /// existing asset index, if any, because guessing one would route orders to
    /// the wrong asset.
    pub fn cache_instrument(&self, instrument: &InstrumentAny) {
        self.merge_instruments(std::slice::from_ref(instrument));
    }

    pub(crate) fn merge_instruments(&self, instruments: &[InstrumentAny]) {
        if instruments.is_empty() {
            return;
        }

        self.cache_asset_indices(instruments);
        self.instruments.rcu(|m| {
            for instrument in instruments {
                m.insert(instrument.symbol().inner(), instrument.clone());
                m.insert(instrument.raw_symbol().inner(), instrument.clone());
            }
        });

        let by_coin: Vec<_> = instruments
            .iter()
            .filter_map(|instrument| {
                let symbol = instrument.symbol();
                match HyperliquidProductType::from_symbol(symbol.as_str()) {
                    Ok(product_type) => Some((instrument, product_type)),
                    Err(_) => {
                        log::warn!("Unable to determine product type for symbol: {symbol}");
                        None
                    }
                }
            })
            .collect();

        if !by_coin.is_empty() {
            self.instruments_by_coin.rcu(|m| {
                for (instrument, product_type) in &by_coin {
                    cache_coin_aliases(m, instrument, *product_type);
                }
            });
        }
    }

    fn cache_asset_indices(&self, instruments: &[InstrumentAny]) {
        let cached = self.asset_indices.load();
        let mut pending = AHashMap::new();

        for instrument in instruments {
            let full_symbol = instrument.symbol().inner();
            let coin = instrument.raw_symbol().inner();
            match instrument
                .info()
                .and_then(|info| info.get_u64(ASSET_INDEX_INFO_KEY))
                .and_then(|value| u32::try_from(value).ok())
            {
                Some(asset_index) => {
                    pending.insert(full_symbol, asset_index);
                }
                // vault tokens are synthesized locally to value balances and are never
                // submitted, so the venue assigns them no asset index to carry
                None if coin.starts_with(VAULT_TOKEN_PREFIX) => {}
                // without an index we cannot address the asset on the wire, so a market we
                // have never indexed is untradable rather than merely stale
                None if !pending.contains_key(&full_symbol)
                    && !cached.contains_key(&full_symbol) =>
                {
                    log::warn!(
                        "Instrument '{full_symbol}' carries no '{ASSET_INDEX_INFO_KEY}' info value \
                         and has no cached asset index; orders for it will be rejected"
                    );
                }
                None => log::warn!(
                    "Instrument '{full_symbol}' carries no '{ASSET_INDEX_INFO_KEY}' info value; \
                     leaving the cached asset index unchanged"
                ),
            }
        }

        drop(cached);

        if !pending.is_empty() {
            self.asset_indices.rcu(|m| {
                m.extend(pending.iter().map(|(symbol, index)| (*symbol, *index)));
            });
        }
    }

    fn get_or_create_instrument(
        &self,
        coin: &Ustr,
        product_type: Option<HyperliquidProductType>,
    ) -> Option<InstrumentAny> {
        if let Some(pt) = product_type
            && let Some(instrument) = self.instruments_by_coin.load().get(&(*coin, pt))
        {
            return Some(instrument.clone());
        }

        // HTTP responses lack product type context. HIP-4 outcome coins
        // (`#E`/`+E`) are checked first because they never collide with
        // perp or spot symbols, then perp, then spot.
        if product_type.is_none() {
            let guard = self.instruments_by_coin.load();

            if let Some(instrument) = guard.get(&(*coin, HyperliquidProductType::Outcome)) {
                return Some(instrument.clone());
            }

            if let Some(instrument) = guard.get(&(*coin, HyperliquidProductType::Perp)) {
                return Some(instrument.clone());
            }

            if let Some(instrument) = guard.get(&(*coin, HyperliquidProductType::Spot)) {
                return Some(instrument.clone());
            }
        }

        // Spot fills use @{pair_index} format, translate to full symbol and look up
        if coin.starts_with('@')
            && let Some(symbol) = self.spot_fill_coins.load().get(coin)
        {
            // Look up by full symbol in instruments map (not instruments_by_coin
            // which uses raw_symbol)
            if let Some(instrument) = self.instruments.load().get(symbol) {
                return Some(instrument.clone());
            }
        }

        // Settled HIP-4 outcomes drop out of `outcomeMeta`, so uncached
        // instruments default to USDC because the encoding lacks a quote token.
        if let Ok(asset_id) = parse_outcome_symbol(coin)
            && let Ok(def) = parse_unlisted_outcome_instrument(asset_id)
            && let Some(instrument) = create_instrument_from_def(&def, self.clock.get_time_ns())
        {
            log::debug!("Creating instrument for unlisted outcome coin: {coin}");
            self.cache_instrument(&instrument);
            return Some(instrument);
        }

        // Vault tokens aren't in standard API, create synthetic instruments
        if coin.starts_with(VAULT_TOKEN_PREFIX) {
            log::debug!("Creating synthetic instrument for vault token: {coin}");

            let ts_event = self.clock.get_time_ns();

            // Create synthetic vault token instrument
            let symbol_str = format!("{coin}-USDC-SPOT");
            let symbol = Symbol::new(&symbol_str);
            let venue = *HYPERLIQUID_VENUE;
            let instrument_id = InstrumentId::new(symbol, venue);

            // Create currencies
            let base_currency = Currency::new(
                coin.as_str(),
                8, // precision
                0, // ISO code (not applicable)
                coin.as_str(),
                CurrencyType::Crypto,
            );

            let quote_currency = Currency::new(
                "USDC",
                6, // USDC standard precision
                0,
                "USDC",
                CurrencyType::Crypto,
            );

            let price_increment = Price::from("0.00000001");
            let size_increment = Quantity::from("0.00000001");

            let instrument = InstrumentAny::CurrencyPair(
                CurrencyPair::builder()
                    .instrument_id(instrument_id)
                    .raw_symbol(symbol)
                    .base_currency(base_currency)
                    .quote_currency(quote_currency)
                    .price_precision(8)
                    .size_precision(8)
                    .price_increment(price_increment)
                    .size_increment(size_increment)
                    .ts_event(ts_event)
                    .ts_init(ts_event)
                    .build()
                    .unwrap(),
            );

            self.cache_instrument(&instrument);

            Some(instrument)
        } else {
            // For non-vault tokens, log warning and return None
            log::warn!("Instrument not found in cache: {coin}");
            None
        }
    }

    /// Set the account ID for this client.
    ///
    /// This is required for generating reports with the correct account ID.
    pub fn set_account_id(&mut self, account_id: AccountId) {
        self.account_id = Some(account_id);
    }

    /// Fetch and parse all instrument definitions, populating the asset indices cache.
    pub async fn request_instrument_defs(&self) -> Result<Vec<HyperliquidInstrumentDef>> {
        Ok(self.load_instrument_defs().await?.0)
    }

    /// Loads all instrument definitions, together with the product types whose
    /// metadata loaded in full.
    async fn load_instrument_defs(
        &self,
    ) -> Result<(
        Vec<HyperliquidInstrumentDef>,
        AHashSet<HyperliquidProductType>,
    )> {
        let mut defs: Vec<HyperliquidInstrumentDef> = Vec::new();
        let mut loaded = AHashSet::new();
        let spot_meta = match self.inner.get_spot_meta().await {
            Ok(spot_meta) => Some(spot_meta),
            Err(e) => {
                log::warn!("Failed to load Hyperliquid spot metadata: {e}");
                None
            }
        };

        // Load all perp dexes: index 0 = standard, index 1+ = HIP-3
        match self.inner.load_all_perp_metas().await {
            Ok(all_metas) => {
                for (dex_index, meta) in all_metas.iter().enumerate() {
                    let base = perp_dex_asset_index_base(dex_index);
                    let settlement_currency = match resolve_perp_settlement_currency(
                        meta,
                        spot_meta.as_ref(),
                    ) {
                        Ok(settlement_currency) => settlement_currency,
                        Err(e) => {
                            return Err(Error::decode(format!(
                                "failed to resolve perp settlement currency for dex {dex_index}: {e}",
                            )));
                        }
                    };

                    let perp_defs = parse_perp_instruments_with_settlement(
                        meta,
                        base,
                        settlement_currency.as_str(),
                    );
                    log::debug!(
                        "Loaded Hyperliquid perp defs: dex_index={dex_index}, count={}",
                        perp_defs.len(),
                    );
                    defs.extend(perp_defs);
                }
                loaded.insert(HyperliquidProductType::Perp);
            }
            Err(e) => {
                log::warn!("Failed to load allPerpMetas, falling back to meta: {e}");

                match self.inner.load_perp_meta().await {
                    Ok(perp_meta) => {
                        match resolve_perp_settlement_currency(&perp_meta, spot_meta.as_ref()) {
                            Ok(settlement_currency) => {
                                let perp_defs = parse_perp_instruments_with_settlement(
                                    &perp_meta,
                                    0,
                                    settlement_currency.as_str(),
                                );
                                log::debug!(
                                    "Loaded Hyperliquid perp defs via fallback: count={}",
                                    perp_defs.len(),
                                );
                                defs.extend(perp_defs);
                            }
                            Err(e) => {
                                return Err(Error::decode(format!(
                                    "failed to resolve fallback perp settlement currency: {e}",
                                )));
                            }
                        }
                    }
                    Err(e) => {
                        log::warn!("Failed to load Hyperliquid perp metadata: {e}");
                    }
                }
            }
        }

        if let Some(spot_meta) = spot_meta.as_ref() {
            match parse_spot_instruments(spot_meta) {
                Ok(spot_defs) => {
                    log::debug!(
                        "Loaded Hyperliquid spot definitions: count={}",
                        spot_defs.len(),
                    );
                    defs.extend(spot_defs);
                    loaded.insert(HyperliquidProductType::Spot);
                }
                Err(e) => {
                    log::warn!("Failed to parse Hyperliquid spot instruments: {e}");
                }
            }
        }

        // HIP-4 outcome metadata is best-effort: the venue may not expose it
        // and the response shape is still firming up. Treat any error as a
        // soft skip so missing outcomes do not break perp/spot loading.
        match self.inner.get_outcome_meta().await {
            Ok(outcome_meta) => match parse_outcome_instruments(&outcome_meta) {
                Ok(outcome_defs) => {
                    log::debug!(
                        "Loaded Hyperliquid outcome definitions: count={}",
                        outcome_defs.len(),
                    );
                    defs.extend(outcome_defs);
                    loaded.insert(HyperliquidProductType::Outcome);
                }
                Err(e) => {
                    log::warn!("Failed to parse Hyperliquid outcome instruments: {e}");
                }
            },
            Err(e) => {
                log::debug!("Skipping Hyperliquid outcome metadata: {e}");
            }
        }

        // Drop defs whose Nautilus-internal symbol collides with one already
        // accepted. This guards the HIP-3 case where two distinct venue names
        // (e.g. `dex:FOO*` and `dex:FOO?`) sanitize onto the same internal
        // symbol; without this filter the second def would silently overwrite
        // the first in `asset_indices`, which would route orders to the wrong
        // asset. First-write-wins matches the spot canonical-pair ordering.
        let mut seen_symbols = ahash::AHashSet::with_capacity(defs.len());
        let mut deduped: Vec<HyperliquidInstrumentDef> = Vec::with_capacity(defs.len());
        for def in defs {
            if seen_symbols.insert(def.symbol) {
                deduped.push(def);
            } else {
                log::warn!(
                    "Dropping Hyperliquid instrument: sanitized symbol '{}' collides with an earlier def (raw_symbol='{}')",
                    def.symbol,
                    def.raw_symbol,
                );
            }
        }
        let defs = deduped;

        // Populate asset indices for all instruments (including filtered HIP-3)
        self.asset_indices.rcu(|m| {
            for def in &defs {
                m.insert(def.symbol, def.asset_index);
            }
        });
        log::debug!(
            "Populated asset indices map (count={})",
            self.asset_indices.len()
        );

        Ok((defs, loaded))
    }

    /// Converts instrument definitions into Nautilus instruments.
    pub fn convert_defs(&self, defs: Vec<HyperliquidInstrumentDef>) -> Vec<InstrumentAny> {
        let ts_init = self.clock.get_time_ns();
        instruments_from_defs_owned(defs, ts_init)
    }

    /// Fetch and parse all available instrument definitions from Hyperliquid.
    pub async fn request_instruments(&self) -> Result<Vec<InstrumentAny>> {
        let defs = self.request_instrument_defs().await?;
        Ok(self.convert_defs(defs))
    }

    /// Fetch all available instruments together with the market status of each.
    pub(crate) async fn request_instruments_with_statuses(&self) -> Result<InstrumentSnapshot> {
        let (defs, loaded) = self.load_instrument_defs().await?;
        let def_statuses: AHashMap<Ustr, MarketStatusAction> = defs
            .iter()
            .map(|def| {
                let action = if def.active {
                    MarketStatusAction::Trading
                } else {
                    MarketStatusAction::Close
                };
                (def.symbol, action)
            })
            .collect();
        let instruments = self.convert_defs(defs);

        // Defs that fail conversion yield no instrument, so they get no status
        let statuses = instruments
            .iter()
            .filter_map(|instrument| {
                let instrument_id = instrument.id();
                def_statuses
                    .get(&instrument_id.symbol.inner())
                    .map(|&action| (instrument_id, action))
            })
            .collect();

        Ok(InstrumentSnapshot {
            instruments,
            statuses,
            loaded,
        })
    }

    /// Builds the `allDexsAssetCtxs` normalization map from dex name to ordered instrument IDs.
    ///
    /// The order of instrument IDs must match the venue universe ordering for each perp dex so
    /// incoming `ctxs` arrays can be normalized without leaking raw positional payloads.
    pub async fn build_all_dex_asset_ctxs_instrument_ids(
        &self,
    ) -> Result<AHashMap<String, Vec<Option<InstrumentId>>>> {
        let all_metas = match self.inner.load_all_perp_metas().await {
            Ok(all_metas) => all_metas,
            Err(e) => {
                log::warn!("Failed to load allPerpMetas, falling back to meta: {e}");
                vec![self.inner.load_perp_meta().await?]
            }
        };

        let perp_dexs = match self.inner.load_perp_dexs().await {
            Ok(dexs) => Some(dexs),
            Err(e) => {
                log::warn!("Failed to load perpDexs, inferring dex names from metadata: {e}");
                None
            }
        };

        let raw_symbol_to_id =
            self.instruments
                .load()
                .values()
                .fold(AHashMap::new(), |mut acc, instrument| {
                    acc.insert(instrument.raw_symbol().to_string(), instrument.id());
                    acc
                });

        let mut mapping = AHashMap::new();

        for (dex_index, meta) in all_metas.iter().enumerate() {
            let dex_name = resolve_perp_dex_name(dex_index, meta, perp_dexs.as_deref());
            let mut instrument_ids = Vec::with_capacity(meta.universe.len());

            for asset in &meta.universe {
                if let Some(instrument_id) = raw_symbol_to_id.get(&asset.name) {
                    instrument_ids.push(Some(*instrument_id));
                } else {
                    log::warn!(
                        "Missing cached Hyperliquid instrument for dex='{}' raw_symbol='{}'",
                        dex_name,
                        asset.name
                    );
                    instrument_ids.push(None);
                }
            }

            mapping.insert(dex_name, instrument_ids);
        }

        Ok(mapping)
    }

    /// Get asset index for a symbol from the cached map.
    ///
    /// For perps: index in meta.universe (0, 1, 2, ...).
    /// For spot: 10_000 + index in spotMeta.universe.
    /// For HIP-3: 100_000 + dex_index * 10_000 + index in dex meta.universe.
    ///
    /// Returns `None` if the symbol is not found in the map.
    pub fn get_asset_index(&self, symbol: &str) -> Option<u32> {
        self.get_asset_index_for_symbol(Ustr::from(symbol))
    }

    /// Get asset index for an already-interned symbol from the cached map.
    ///
    /// Returns `None` if the symbol is not found in the map.
    pub(crate) fn get_asset_index_for_symbol(&self, symbol: Ustr) -> Option<u32> {
        self.asset_indices.load().get(&symbol).copied()
    }

    /// Get the price precision for a cached instrument by symbol.
    pub fn get_price_precision(&self, symbol: &str) -> Option<u8> {
        self.get_price_precision_for_symbol(Ustr::from(symbol))
    }

    /// Get the price precision for a cached instrument by interned symbol.
    pub(crate) fn get_price_precision_for_symbol(&self, symbol: Ustr) -> Option<u8> {
        self.instruments
            .load()
            .get(&symbol)
            .map(|inst| inst.price_precision())
    }

    /// Get mapping from spot fill coin identifiers to instrument symbols.
    ///
    /// Hyperliquid WebSocket fills for spot use `@{pair_index}` format (e.g., `@107`),
    /// while instruments are identified by full symbols (e.g., `HYPE-USDC-SPOT`).
    /// This mapping allows looking up the instrument from a spot fill.
    ///
    /// This method also caches the mapping internally for use by fill parsing methods.
    #[must_use]
    pub fn get_spot_fill_coin_mapping(&self) -> AHashMap<Ustr, Ustr> {
        const SPOT_INDEX_OFFSET: u32 = 10_000;
        const BUILDER_PERP_OFFSET: u32 = 100_000;

        let guard = self.asset_indices.load();

        let mut mapping = AHashMap::new();

        for (symbol, &asset_index) in guard.iter() {
            // Spot instruments: asset_index in [10_000, 100_000)
            if (SPOT_INDEX_OFFSET..BUILDER_PERP_OFFSET).contains(&asset_index) {
                let pair_index = asset_index - SPOT_INDEX_OFFSET;
                let fill_coin = Ustr::from(&format!("@{pair_index}"));
                mapping.insert(fill_coin, *symbol);
            }
        }

        // Cache the mapping internally for fill parsing
        self.spot_fill_coins.store(mapping.clone());

        mapping
    }

    /// Gets perpetuals metadata for internal use.
    #[allow(dead_code)]
    pub(crate) async fn load_perp_meta(&self) -> Result<PerpMeta> {
        self.inner.load_perp_meta().await
    }

    /// Get metadata for all perp dexes (standard + HIP-3).
    #[allow(dead_code)]
    pub(crate) async fn load_all_perp_metas(&self) -> Result<Vec<PerpMeta>> {
        self.inner.load_all_perp_metas().await
    }

    /// Gets spot metadata for internal use.
    #[allow(dead_code)]
    pub(crate) async fn get_spot_meta(&self) -> Result<SpotMeta> {
        self.inner.get_spot_meta().await
    }

    /// Gets outcome metadata for internal use.
    pub(crate) async fn get_outcome_meta(&self) -> Result<OutcomeMeta> {
        self.inner.get_outcome_meta().await
    }

    pub(crate) fn outcome_quote_currency(
        &self,
        instrument_id: InstrumentId,
        outcome_index: u32,
        meta: &OutcomeMeta,
    ) -> Currency {
        if let Some(instrument) = self.instruments.load().get(&instrument_id.symbol.inner()) {
            return instrument.quote_currency();
        }

        let code = meta
            .outcomes
            .iter()
            .find(|market| market.outcome == outcome_index)
            .and_then(|market| market.quote_token.as_deref())
            .unwrap_or(DEFAULT_OUTCOME_QUOTE_CURRENCY);
        get_outcome_currency(code)
    }

    /// Get L2 order book for a coin.
    pub async fn info_l2_book(&self, coin: &str) -> Result<HyperliquidL2Book> {
        self.inner.info_l2_book(coin).await
    }

    /// Get recent public trades for a coin.
    pub async fn info_recent_trades(&self, coin: &str) -> Result<Vec<HyperliquidRecentTrade>> {
        self.inner.info_recent_trades(coin).await
    }

    /// Get user fills (trading history).
    pub async fn info_user_fills(&self, user: &str) -> Result<HyperliquidFills> {
        self.inner.info_user_fills(user).await
    }

    /// Get order status for a user.
    pub async fn info_order_status(&self, user: &str, oid: u64) -> Result<HyperliquidOrderStatus> {
        self.inner.info_order_status(user, oid).await
    }

    /// Get all open orders for a user.
    pub async fn info_open_orders(&self, user: &str) -> Result<Value> {
        self.inner.info_open_orders(user).await
    }

    /// Get frontend open orders (includes more detail) for a user.
    pub async fn info_frontend_open_orders(&self, user: &str) -> Result<Value> {
        self.inner.info_frontend_open_orders(user).await
    }

    async fn info_frontend_open_orders_for_dex(
        &self,
        user: &str,
        dex: Option<&str>,
    ) -> Result<Value> {
        self.inner
            .info_frontend_open_orders_for_dex(user, dex)
            .await
    }

    /// Get the most recent historical orders for a user.
    pub async fn info_historical_orders(
        &self,
        user: &str,
    ) -> Result<Vec<HyperliquidOrderStatusEntry>> {
        self.inner.info_historical_orders(user).await
    }

    /// Get clearinghouse state (balances, positions, margin) for a user.
    pub async fn info_clearinghouse_state(&self, user: &str) -> Result<Value> {
        self.inner.info_clearinghouse_state(user).await
    }

    async fn info_clearinghouse_state_for_dex(
        &self,
        user: &str,
        dex: Option<&str>,
    ) -> Result<Value> {
        self.inner.info_clearinghouse_state_for_dex(user, dex).await
    }

    /// Get spot clearinghouse state (per-token spot balances) for a user.
    pub async fn info_spot_clearinghouse_state(&self, user: &str) -> Result<Value> {
        self.inner.info_spot_clearinghouse_state(user).await
    }

    /// Get the account abstraction mode for a user.
    ///
    /// A mode string this adapter does not recognize maps to
    /// [`HyperliquidAccountAbstraction::Unknown`] and is logged as a warning; such accounts keep
    /// the perp-summary balance logic.
    ///
    /// # Errors
    ///
    /// Returns an error if the request fails or the response is not a JSON string.
    pub async fn info_user_abstraction(&self, user: &str) -> Result<HyperliquidAccountAbstraction> {
        let response = self.inner.info_user_abstraction(user).await?;

        // Serde reads a single-key object like `{"mode": null}` as a unit variant, which
        // `#[serde(other)]` would map to `Unknown`, so only a bare string is accepted. Any
        // string decodes, to a known mode or to `Unknown`
        let abstraction = response
            .as_str()
            .and_then(|_| serde_json::from_value(response.clone()).ok())
            .ok_or_else(|| {
                Error::decode(format!(
                    "Failed to parse user abstraction: expected a mode string, was {response}"
                ))
            })?;

        if abstraction == HyperliquidAccountAbstraction::Unknown {
            log::warn!(
                "Unrecognized Hyperliquid account abstraction {response}, using perp clearinghouse balances"
            );
        } else {
            log::debug!("Hyperliquid account abstraction: {abstraction}");
        }

        Ok(abstraction)
    }

    /// Get user fee schedule and effective rates.
    pub async fn info_user_fees(&self, user: &str) -> Result<Value> {
        self.inner.info_user_fees(user).await
    }

    /// Get candle/bar data for a coin.
    pub async fn info_candle_snapshot(
        &self,
        coin: &str,
        interval: HyperliquidBarInterval,
        start_time: u64,
        end_time: u64,
    ) -> Result<HyperliquidCandleSnapshot> {
        self.inner
            .info_candle_snapshot(coin, interval, start_time, end_time)
            .await
    }

    /// Get historical funding rates for a coin.
    pub async fn info_funding_history(
        &self,
        coin: &str,
        start_time: u64,
        end_time: Option<u64>,
    ) -> Result<Vec<HyperliquidFundingHistoryEntry>> {
        self.inner
            .info_funding_history(coin, start_time, end_time)
            .await
    }

    /// Post an action to the exchange endpoint (low-level delegation).
    pub async fn post_action(
        &self,
        action: &ExchangeAction,
    ) -> Result<HyperliquidExchangeResponse> {
        self.inner.post_action(action).await
    }

    /// Post an execution action (low-level delegation).
    pub async fn post_action_exec(
        &self,
        action: &HyperliquidExchangeAction,
    ) -> Result<HyperliquidExchangeResponse> {
        self.inner.post_action_exec(action).await
    }

    /// Build the signed exchange request used by both HTTP and WebSocket post transports.
    ///
    /// # Errors
    ///
    /// Returns an error if the action is unsupported, credentials are missing, or signing fails.
    pub fn sign_action_exec_request(
        &self,
        action: &HyperliquidExchangeAction,
        expires_after: Option<u64>,
    ) -> Result<HyperliquidExchangeRequest<HyperliquidExchangeAction>> {
        self.inner.sign_action_exec_request(action, expires_after)
    }

    /// Get metadata about available markets (low-level delegation).
    pub async fn info_meta(&self) -> Result<HyperliquidMeta> {
        self.inner.info_meta().await
    }

    /// Cancel an order on the Hyperliquid exchange.
    ///
    /// Can cancel either by venue order ID or client order ID.
    /// At least one ID must be provided.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, no order ID is provided,
    /// or the API returns an error.
    pub async fn cancel_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: Option<VenueOrderId>,
    ) -> Result<()> {
        // Get asset ID from cached indices map
        let symbol = instrument_id.symbol.inner();
        let asset_id = self.get_asset_index_for_symbol(symbol).ok_or_else(|| {
            Error::bad_request(format!(
                "Asset index not found for symbol: {symbol}. Ensure instruments are loaded."
            ))
        })?;

        let action = if let Some(client_order_id) = client_order_id {
            if let Some(cloid) = self.cached_client_order_id_cloid(&client_order_id) {
                HyperliquidExchangeAction::CancelByCloid {
                    cancels: vec![HyperliquidExchangeCancelByCloidRequest {
                        asset: asset_id,
                        cloid,
                    }],
                    fast: None,
                }
            } else if let Some(oid) = venue_order_id {
                let oid_u64 = oid
                    .as_str()
                    .parse::<u64>()
                    .map_err(|_| Error::bad_request("Invalid venue order ID format"))?;
                HyperliquidExchangeAction::Cancel {
                    cancels: vec![HyperliquidExchangeCancelOrderRequest {
                        asset: asset_id,
                        oid: oid_u64,
                    }],
                    fast: None,
                }
            } else {
                let cloid = self.get_or_generate_client_order_id_cloid(client_order_id);
                HyperliquidExchangeAction::CancelByCloid {
                    cancels: vec![HyperliquidExchangeCancelByCloidRequest {
                        asset: asset_id,
                        cloid,
                    }],
                    fast: None,
                }
            }
        } else if let Some(oid) = venue_order_id {
            let oid_u64 = oid
                .as_str()
                .parse::<u64>()
                .map_err(|_| Error::bad_request("Invalid venue order ID format"))?;
            HyperliquidExchangeAction::Cancel {
                cancels: vec![HyperliquidExchangeCancelOrderRequest {
                    asset: asset_id,
                    oid: oid_u64,
                }],
                fast: None,
            }
        } else {
            return Err(Error::bad_request(
                "Either client_order_id or venue_order_id must be provided",
            ));
        };

        // Submit cancellation
        let response = self.inner.post_action_exec(&action).await?;

        // Check response - only check for error status
        match response {
            ref r @ HyperliquidExchangeResponse::Status { .. } if r.is_ok() => Ok(()),
            HyperliquidExchangeResponse::Status {
                status,
                response: error_data,
            } => Err(Error::bad_request(format!(
                "Cancel order failed: status={status}, error={error_data}"
            ))),
            HyperliquidExchangeResponse::Error { error } => {
                Err(Error::bad_request(format!("Cancel order error: {error}")))
            }
        }
    }

    /// Modify an order on the Hyperliquid exchange.
    ///
    /// The HL modify API requires a full replacement order spec plus a venue
    /// order ID or cached CLOID target. The caller must provide all order fields.
    ///
    /// # Errors
    ///
    /// Returns an error if the asset index is not found, no safe modify target
    /// exists, the venue order ID is invalid, or the API returns an error.
    #[expect(clippy::too_many_arguments)]
    pub async fn modify_order(
        &self,
        instrument_id: InstrumentId,
        venue_order_id: Option<VenueOrderId>,
        order_side: OrderSide,
        order_type: OrderType,
        price: Price,
        quantity: Quantity,
        trigger_price: Option<Price>,
        reduce_only: bool,
        post_only: bool,
        time_in_force: TimeInForce,
        client_order_id: Option<ClientOrderId>,
    ) -> Result<()> {
        let symbol = instrument_id.symbol.inner();
        let asset_id = self.get_asset_index_for_symbol(symbol).ok_or_else(|| {
            Error::bad_request(format!(
                "Asset index not found for symbol: {symbol}. Ensure instruments are loaded."
            ))
        })?;

        let oid = match client_order_id
            .as_ref()
            .and_then(|id| self.unique_cached_client_order_id_cloid(id))
        {
            Some(cloid) => HyperliquidExchangeModifyTarget::Cloid(cloid),
            None => {
                let Some(venue_order_id) = venue_order_id.as_ref() else {
                    return Err(Error::bad_request(
                        "venue_order_id or unique cached CLOID is required for modify",
                    ));
                };
                HyperliquidExchangeModifyTarget::from_venue_order_id(venue_order_id)
                    .map_err(|_| Error::bad_request("Invalid venue order ID format"))?
            }
        };

        let is_buy = matches!(order_side, OrderSide::Buy);
        let decimals = self.get_price_precision_for_symbol(symbol);

        let normalized_price = normalize_or_validate_wire_price(
            price.as_decimal(),
            "Price",
            decimals,
            self.normalize_prices,
        )
        .map_err(|e| Error::bad_request(format!("{e}")))?;

        let size = quantity.as_decimal().normalize();

        let kind = match order_type {
            OrderType::Market => HyperliquidExchangeOrderKind::Limit {
                limit: HyperliquidExchangeLimitParams {
                    tif: HyperliquidExchangeTif::Ioc,
                },
            },
            OrderType::Limit => {
                let tif = time_in_force_to_hyperliquid_tif(time_in_force, post_only)
                    .map_err(|e| Error::bad_request(format!("{e}")))?;
                HyperliquidExchangeOrderKind::Limit {
                    limit: HyperliquidExchangeLimitParams { tif },
                }
            }
            OrderType::StopMarket
            | OrderType::StopLimit
            | OrderType::MarketIfTouched
            | OrderType::LimitIfTouched => {
                if let Some(trig_px) = trigger_price {
                    let trigger_price_decimal = normalize_or_validate_wire_price(
                        trig_px.as_decimal(),
                        "Trigger price",
                        decimals,
                        self.normalize_prices,
                    )
                    .map_err(|e| Error::bad_request(format!("{e}")))?;
                    let tpsl = match order_type {
                        OrderType::StopMarket | OrderType::StopLimit => HyperliquidExchangeTpSl::Sl,
                        _ => HyperliquidExchangeTpSl::Tp,
                    };
                    let is_market = matches!(
                        order_type,
                        OrderType::StopMarket | OrderType::MarketIfTouched
                    );
                    HyperliquidExchangeOrderKind::Trigger {
                        trigger: HyperliquidExchangeTriggerParams {
                            is_market,
                            trigger_px: trigger_price_decimal,
                            tpsl,
                        },
                    }
                } else {
                    return Err(Error::bad_request("Trigger orders require a trigger price"));
                }
            }
            _ => {
                return Err(Error::bad_request(format!(
                    "Order type {order_type:?} not supported for modify"
                )));
            }
        };
        let cloid = client_order_id.map(|id| self.get_or_generate_client_order_id_cloid(id));

        let order = HyperliquidExchangePlaceOrderRequest {
            asset: asset_id,
            is_buy,
            price: normalized_price,
            size,
            reduce_only,
            kind,
            cloid,
        };

        let action = HyperliquidExchangeAction::Modify {
            modify: HyperliquidExchangeModifyOrderRequest { oid, order },
        };

        let response = self.inner.post_action_exec(&action).await?;

        match response {
            ref r @ HyperliquidExchangeResponse::Status { .. } if r.is_ok() => {
                if let Some(inner_error) = extract_inner_error(&response) {
                    Err(Error::bad_request(format!(
                        "Modify order rejected: {inner_error}",
                    )))
                } else {
                    Ok(())
                }
            }
            HyperliquidExchangeResponse::Status {
                status,
                response: error_data,
            } => Err(Error::bad_request(format!(
                "Modify order failed: status={status}, error={error_data}"
            ))),
            HyperliquidExchangeResponse::Error { error } => {
                Err(Error::bad_request(format!("Modify order error: {error}")))
            }
        }
    }

    /// Arms, replaces, or clears the account-wide scheduled cancellation of all open orders.
    ///
    /// `time` is a Unix timestamp in milliseconds, at least five seconds after the local
    /// clock when called. `None` clears the schedule. Allow extra time for rate limiting,
    /// network latency, and clock skew. The venue can still reject a delayed request.
    /// The schedule remains at the venue after disconnect or process exit; this method
    /// does not start a refresh loop or close positions.
    /// Convert Nautilus nanosecond timestamps to milliseconds before passing them here.
    /// The venue requires $1,000,000 trading volume and permits at most ten schedule triggers
    /// per day, resetting the trigger count at 00:00 UTC.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The timestamp is less than five seconds ahead of the local clock.
    /// - Credentials are missing or the request fails.
    /// - The venue rejects the action or returns an unexpected acknowledgement.
    pub async fn schedule_cancel(&self, time: Option<u64>) -> Result<()> {
        if let Some(time) = time {
            validate_schedule_cancel_time(time, self.clock.get_time_ms())?;
        }

        let action = HyperliquidExchangeAction::ScheduleCancel { time };
        let response = self.inner.post_action_exec(&action).await?;
        match response {
            HyperliquidExchangeResponse::Status { status, response }
                if status == RESPONSE_STATUS_OK =>
            {
                let data = serde_json::from_value::<HyperliquidExchangeResponseData>(response)
                    .map_err(Error::Serde)?;

                if matches!(data, HyperliquidExchangeResponseData::Default) {
                    Ok(())
                } else {
                    Err(Error::bad_request(
                        "Unexpected schedule cancel acknowledgement",
                    ))
                }
            }
            response => Err(Error::bad_request(format!(
                "Schedule cancel failed: {response:?}"
            ))),
        }
    }

    /// Split an HIP-4 outcome's quote tokens into matched Yes and No side tokens.
    ///
    /// Submits a `userOutcome` exchange action with the `splitOutcome` operation:
    /// debits `amount` quote tokens and credits `amount` Yes plus `amount`
    /// No side tokens for the given `outcome` index. Ordinary directional
    /// buys and sells on outcome instruments go through the standard order path
    /// without calling this; the action is for dual-side market making and
    /// inventory creation.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the venue rejects the
    /// action, or the response cannot be parsed.
    pub async fn submit_split_outcome(
        &self,
        outcome: u32,
        amount: Decimal,
    ) -> Result<HyperliquidExchangeResponse> {
        let action = HyperliquidExchangeAction::UserOutcome {
            op: HyperliquidExchangeUserOutcomeOp::SplitOutcome(
                HyperliquidExchangeSplitOutcomeParams { outcome, amount },
            ),
        };
        self.inner.post_action_exec(&action).await
    }

    /// Merge matched Yes + No side-token pairs of an HIP-4 outcome back into quote tokens.
    ///
    /// Submits a `userOutcome` action with the `mergeOutcome` operation. Pass
    /// `amount = None` to merge the maximum mergeable balance (venue-side
    /// `null`).
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the venue rejects the
    /// action, or the response cannot be parsed.
    pub async fn submit_merge_outcome(
        &self,
        outcome: u32,
        amount: Option<Decimal>,
    ) -> Result<HyperliquidExchangeResponse> {
        let action = HyperliquidExchangeAction::UserOutcome {
            op: HyperliquidExchangeUserOutcomeOp::MergeOutcome(
                HyperliquidExchangeMergeOutcomeParams { outcome, amount },
            ),
        };
        self.inner.post_action_exec(&action).await
    }

    /// Merge `Yes` shares of every outcome in a multi-outcome question into quote tokens.
    ///
    /// Submits a `userOutcome` action with the `mergeQuestion` operation. Pass
    /// `amount = None` to merge the maximum balance.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the venue rejects the
    /// action, or the response cannot be parsed.
    pub async fn submit_merge_question(
        &self,
        question: u32,
        amount: Option<Decimal>,
    ) -> Result<HyperliquidExchangeResponse> {
        let action = HyperliquidExchangeAction::UserOutcome {
            op: HyperliquidExchangeUserOutcomeOp::MergeQuestion(
                HyperliquidExchangeMergeQuestionParams { question, amount },
            ),
        };
        self.inner.post_action_exec(&action).await
    }

    /// Swap `No` shares of one outcome into `Yes` shares of every other outcome.
    ///
    /// Submits a `userOutcome` action with the `negateOutcome` operation. Both
    /// outcomes must belong to the same multi-outcome `question`.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, the venue rejects the
    /// action, or the response cannot be parsed.
    pub async fn submit_negate_outcome(
        &self,
        question: u32,
        outcome: u32,
        amount: Decimal,
    ) -> Result<HyperliquidExchangeResponse> {
        let action = HyperliquidExchangeAction::UserOutcome {
            op: HyperliquidExchangeUserOutcomeOp::NegateOutcome(
                HyperliquidExchangeNegateOutcomeParams {
                    question,
                    outcome,
                    amount,
                },
            ),
        };
        self.inner.post_action_exec(&action).await
    }

    /// Request order status reports for a user.
    ///
    /// Fetches frontend open orders from the default and all cached builder dexes when unfiltered,
    /// or from the dex selected by an instrument filter, then parses them into OrderStatusReports.
    /// This method requires instruments to be added to the client cache via `cache_instrument()`.
    ///
    /// For vault tokens (starting with "vntls:") that are not in the cache, synthetic instruments
    /// will be created automatically.
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails, parsing fails, or a venue row cannot be resolved
    /// to an instrument or converted into a report (the snapshot is then incomplete and must not be
    /// treated as authoritative).
    pub async fn request_order_status_reports(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<Vec<OrderStatusReport>> {
        let dexes = self.reconciliation_dexes(instrument_id);
        let sweep = self
            .request_order_status_reports_for_dexes(user, instrument_id, &dexes)
            .await?;

        if !sweep.complete {
            return Err(Error::bad_request(
                "Open-order snapshot incomplete: at least one venue row could not be decoded, resolved to an instrument, or converted into a report",
            ));
        }

        Ok(sweep.reports)
    }

    pub(crate) async fn request_order_status_reports_for_dexes(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
        dexes: &[Option<Ustr>],
    ) -> Result<ReportSweep<OrderStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let mut reports = Vec::new();
        let mut complete = true;
        let ts_init = self.clock.get_time_ns();

        for dex in dexes {
            let response = self
                .info_frontend_open_orders_for_dex(user, dex.as_deref())
                .await?;
            let orders: Vec<serde_json::Value> = serde_json::from_value(response)
                .map_err(|e| Error::bad_request(format!("Failed to parse orders: {e}")))?;

            for order_value in orders {
                let order: WsBasicOrderData = match serde_json::from_value(order_value) {
                    Ok(order) => order,
                    Err(e) => {
                        log::warn!("Failed to parse order: {e}");
                        complete = false;
                        continue;
                    }
                };

                let instrument = match self.get_or_create_instrument(&order.coin, None) {
                    Some(instrument) => instrument,
                    // get_or_create_instrument warns with the coin
                    None => {
                        complete = false;
                        continue;
                    }
                };

                if instrument_id.is_some_and(|filter_id| instrument.id() != filter_id) {
                    continue;
                }

                match parse_order_status_report_from_basic(
                    &order,
                    &HyperliquidOrderStatusEnum::Open,
                    &instrument,
                    account_id,
                    ts_init,
                ) {
                    Ok(report) => reports.push(report),
                    Err(e) => {
                        log::error!("Failed to parse order status report: {e}");
                        complete = false;
                    }
                }
            }
        }

        Ok(ReportSweep { reports, complete })
    }

    /// Request historical order status reports for a user.
    ///
    /// The venue bounds this endpoint to its 2,000 most recent historical
    /// orders. Mass-status reconciliation narrows these reports to venue order
    /// IDs represented by the retained fill window.
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails or a venue row cannot be resolved to an
    /// instrument or converted into a report (the snapshot is then incomplete and must not be
    /// treated as authoritative).
    pub async fn request_historical_order_status_reports(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<Vec<OrderStatusReport>> {
        let entries = self.info_historical_orders(user).await?;
        let sweep = self.historical_order_status_reports_from_response(entries, instrument_id)?;

        if !sweep.complete {
            return Err(Error::bad_request(
                "Historical-order snapshot incomplete: at least one venue row could not be decoded, resolved to an instrument, or converted into a report",
            ));
        }

        Ok(sweep.reports)
    }

    pub(crate) fn historical_order_status_reports_from_response(
        &self,
        entries: Vec<HyperliquidOrderStatusEntry>,
        instrument_id: Option<InstrumentId>,
    ) -> Result<ReportSweep<OrderStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let mut reports = Vec::new();
        let mut complete = true;
        let ts_init = self.clock.get_time_ns();

        for entry in entries {
            let instrument = match self.get_or_create_instrument(&entry.order.coin, None) {
                // get_or_create_instrument warns with the coin
                Some(instrument) => instrument,
                None => {
                    complete = false;
                    continue;
                }
            };

            if instrument_id.is_some_and(|filter_id| instrument.id() != filter_id) {
                continue;
            }

            let label = entry.order.order_type.as_deref().unwrap_or_default();
            let tpsl = parse_trigger_order_type_label(label).map(|(tpsl, _)| tpsl);
            let is_market = label.ends_with("Market");
            let historical_order_type = match tpsl.as_ref() {
                Some(tpsl) => parse_trigger_order_type(is_market, tpsl),
                None if is_market => OrderType::Market,
                None => OrderType::Limit,
            };
            let order = WsBasicOrderData {
                coin: entry.order.coin,
                side: entry.order.side,
                limit_px: entry.order.limit_px,
                sz: entry.order.sz,
                oid: entry.order.oid,
                timestamp: entry.order.timestamp,
                orig_sz: entry.order.orig_sz,
                cloid: entry.order.cloid,
                tif: entry.order.tif,
                reduce_only: entry.order.reduce_only,
                trigger_px: entry
                    .order
                    .trigger_px
                    .filter(|price| *price != Decimal::ZERO),
                is_market: tpsl.is_some().then_some(is_market),
                tpsl,
                trigger_activated: None,
                trailing_stop: None,
                order_type: None,
            };

            match parse_order_status_report_from_basic(
                &order,
                &entry.status,
                &instrument,
                account_id,
                ts_init,
            ) {
                Ok(mut report) => {
                    report.order_type = historical_order_type;
                    report.ts_last = UnixNanos::from(entry.status_timestamp * 1_000_000);
                    reports.push(report);
                }
                Err(e) => {
                    log::error!("Failed to parse historical order status report: {e}");
                    complete = false;
                }
            }
        }

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

    /// Request a single order status report by venue order ID.
    ///
    /// Queries `info_frontend_open_orders` and filters for the given oid so the
    /// result includes trigger metadata (trigger_px, tpsl, trailing_stop, etc.).
    /// Falls back to `info_order_status` when the order is no longer open.
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails, parsing fails, or the matched venue row cannot be
    /// resolved to an instrument or converted into a report. A genuinely absent order returns
    /// `Ok(None)`.
    pub async fn request_order_status_report(
        &self,
        user: &str,
        oid: u64,
    ) -> Result<Option<OrderStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;

        let ts_init = self.clock.get_time_ns();

        // Try open orders first (returns full WsBasicOrderData with trigger fields).
        // A transport error here must not abort the call: the oid fallback to
        // info_order_status below still covers closed orders, so a transient
        // frontendOpenOrders outage is downgraded to a warning.
        let orders: Vec<WsBasicOrderData> = match self.info_frontend_open_orders(user).await {
            Ok(response) => match serde_json::from_value(response) {
                Ok(v) => v,
                Err(e) => {
                    log::warn!("Failed to parse frontend open orders response: {e}");
                    Vec::new()
                }
            },
            Err(e) => {
                log::warn!(
                    "Failed to fetch frontendOpenOrders for oid {oid}: {e}; falling back to orderStatus"
                );
                Vec::new()
            }
        };

        if let Some(order) = orders.into_iter().find(|o| o.oid == oid) {
            let instrument = match self.get_or_create_instrument(&order.coin, None) {
                Some(inst) => inst,
                None => {
                    return Err(Error::bad_request(format!(
                        "Failed to resolve instrument for open order oid {oid} with coin {}",
                        order.coin,
                    )));
                }
            };

            let status = if order.trigger_activated == Some(true) {
                HyperliquidOrderStatusEnum::Triggered
            } else {
                HyperliquidOrderStatusEnum::Open
            };

            return parse_order_status_report_from_basic(
                &order,
                &status,
                &instrument,
                account_id,
                ts_init,
            )
            .map(Some)
            .map_err(|e| {
                Error::bad_request(format!(
                    "Failed to parse order status report for oid {oid}: {e}"
                ))
            });
        }

        // Order not in open set: query by oid (returns limited HyperliquidOrderInfo)
        let response = self.info_order_status(user, oid).await?;
        let entry = match response.into_order() {
            Some(e) => e,
            None => return Ok(None),
        };

        let instrument = match self.get_or_create_instrument(&entry.order.coin, None) {
            Some(inst) => inst,
            None => {
                return Err(Error::bad_request(format!(
                    "Failed to resolve instrument for order oid {oid} with coin {}",
                    entry.order.coin,
                )));
            }
        };

        // The info_order_status endpoint returns limited HyperliquidOrderInfo
        // without trigger fields (trigger_px, tpsl, is_market, trailing_stop).
        // Closed trigger orders will report as Limit type. This is an exchange
        // API limitation: trigger metadata is only available on open orders.
        // The reduce-only flag is carried through so reconciliation can clamp a
        // reduce-only fill to its fills.
        let basic = WsBasicOrderData {
            coin: entry.order.coin,
            side: entry.order.side,
            limit_px: entry.order.limit_px,
            sz: entry.order.sz,
            oid: entry.order.oid,
            timestamp: entry.order.timestamp,
            orig_sz: entry.order.orig_sz,
            cloid: entry.order.cloid,
            tif: None,
            reduce_only: entry.order.reduce_only,
            trigger_px: None,
            is_market: None,
            tpsl: None,
            trigger_activated: None,
            trailing_stop: None,
            order_type: None,
        };

        let mut report = parse_order_status_report_from_basic(
            &basic,
            &entry.status,
            &instrument,
            account_id,
            ts_init,
        )
        .map_err(|e| {
            Error::bad_request(format!(
                "Failed to parse order status report for oid {oid}: {e}"
            ))
        })?;

        // Use status_timestamp for ts_last when available (more accurate
        // than the order creation timestamp for filled/canceled orders)
        if entry.status_timestamp > 0 {
            report.ts_last = UnixNanos::from(entry.status_timestamp * 1_000_000);
        }

        Ok(Some(report))
    }

    /// Request a single order status report by client order ID.
    ///
    /// Searches `info_frontend_open_orders` for an order whose cloid matches the
    /// cached CLOID or the generated CLOID. Only finds open orders.
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails, the response cannot be decoded, or the matched
    /// venue row cannot be resolved to an instrument or converted into a report. A genuinely
    /// absent order returns `Ok(None)`.
    pub async fn request_order_status_report_by_client_order_id(
        &self,
        user: &str,
        client_order_id: &ClientOrderId,
    ) -> Result<Option<OrderStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;

        let ts_init = self.clock.get_time_ns();

        let cached_cloid_hex = self
            .cached_client_order_id_cloid(client_order_id)
            .map(|cloid| cloid.to_hex());
        let cloid = Cloid::from_client_order_id(*client_order_id);
        let cloid_hex = cloid.to_hex();

        let response = self.info_frontend_open_orders(user).await?;

        let orders: Vec<WsBasicOrderData> = serde_json::from_value(response).map_err(|e| {
            Error::bad_request(format!("Failed to parse open orders response: {e}"))
        })?;

        let order = match orders.into_iter().find(|o| {
            o.cloid
                .as_ref()
                .is_some_and(|c| cached_cloid_hex.as_ref() == Some(c) || c == &cloid_hex)
        }) {
            Some(o) => o,
            None => return Ok(None),
        };

        let instrument = match self.get_or_create_instrument(&order.coin, None) {
            Some(inst) => inst,
            None => {
                return Err(Error::bad_request(format!(
                    "Failed to resolve instrument for open order with cloid {cloid_hex} and coin {}",
                    order.coin,
                )));
            }
        };

        let status = if order.trigger_activated == Some(true) {
            HyperliquidOrderStatusEnum::Triggered
        } else {
            HyperliquidOrderStatusEnum::Open
        };

        let mut report =
            parse_order_status_report_from_basic(&order, &status, &instrument, account_id, ts_init)
                .map_err(|e| {
                    Error::bad_request(format!(
                        "Failed to parse order status report for cloid {cloid_hex}: {e}"
                    ))
                })?;

        report.client_order_id = Some(*client_order_id);
        Ok(Some(report))
    }

    /// Request fill reports for a user.
    ///
    /// Fetches user fills via `info_user_fills` and parses them into FillReports.
    /// This method requires instruments to be added to the client cache via `cache_instrument()`.
    ///
    /// For vault tokens (starting with "vntls:") that are not in the cache, synthetic instruments
    /// will be created automatically.
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails, parsing fails, or a venue row cannot be resolved
    /// to an instrument or converted into a report (the snapshot is then incomplete and must not be
    /// treated as authoritative).
    ///
    /// Returns an error if `account_id` is not set on the client.
    pub async fn request_fill_reports(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<Vec<FillReport>> {
        let fills_response = self.info_user_fills(user).await?;
        let sweep = self.fill_reports_from_response(fills_response, instrument_id)?;

        if !sweep.complete {
            return Err(Error::bad_request(
                "Fill snapshot incomplete: at least one venue row could not be decoded, resolved to an instrument, or converted into a report",
            ));
        }

        Ok(sweep.reports)
    }

    pub(crate) fn fill_reports_from_response(
        &self,
        fills_response: HyperliquidFills,
        instrument_id: Option<InstrumentId>,
    ) -> Result<ReportSweep<FillReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;

        let mut reports = Vec::new();
        let mut complete = true;
        let ts_init = self.clock.get_time_ns();

        for fill in fills_response {
            // Get instrument from cache or create synthetic for vault tokens
            let instrument = match self.get_or_create_instrument(&fill.coin, None) {
                Some(inst) => inst,
                // get_or_create_instrument warns with the coin
                None => {
                    complete = false;
                    continue;
                }
            };

            // Filter by instrument_id if specified
            if let Some(filter_id) = instrument_id
                && instrument.id() != filter_id
            {
                continue;
            }

            // Parse to FillReport
            match parse_fill_report(&fill, &instrument, account_id, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    log::error!("Failed to parse fill report: {e}");
                    complete = false;
                }
            }
        }

        Ok(ReportSweep { reports, complete })
    }

    /// Request position status reports for a user.
    ///
    /// Fetches clearinghouse state from the default and all cached builder dexes when unfiltered,
    /// plus spot clearinghouse state, then returns the union of perp asset positions (short/long
    /// with PnL) and spot holdings (long only). This method requires instruments to be added to the
    /// client cache via `cache_instrument()`.
    ///
    /// When `instrument_id` resolves to a specific product type, the opposite
    /// product's endpoint is skipped to avoid wasted round trips and make
    /// filtered queries independent of the unused endpoint's availability.
    /// HIP-4 outcomes live in `spotClearinghouseState`, so an outcome filter
    /// is routed like a spot filter (perp leg skipped).
    ///
    /// For vault tokens (starting with "vntls:") that are not in the cache,
    /// synthetic instruments will be created automatically.
    ///
    /// # Errors
    ///
    /// Returns an error if any clearinghouse request fails (when that product or dex is in scope),
    /// parsing fails, or a venue row cannot be resolved to an instrument or converted into a
    /// report (the snapshot is then incomplete and must not be treated as authoritative).
    ///
    /// Returns an error if `account_id` has not been set on the client.
    pub async fn request_position_status_reports(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<Vec<PositionStatusReport>> {
        let dexes = self.reconciliation_dexes(instrument_id);
        let sweep = self
            .request_position_status_reports_for_dexes(user, instrument_id, &dexes)
            .await?;

        if !sweep.complete {
            return Err(Error::bad_request(
                "Position snapshot incomplete: at least one venue row could not be decoded, resolved to an instrument, or converted into a report",
            ));
        }

        Ok(sweep.reports)
    }

    pub(crate) async fn request_position_status_reports_for_dexes(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
        dexes: &[Option<Ustr>],
    ) -> Result<ReportSweep<PositionStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;

        let filter_product = instrument_id
            .and_then(|id| HyperliquidProductType::from_symbol(id.symbol.as_str()).ok());

        let fetch_perp = !matches!(
            filter_product,
            Some(HyperliquidProductType::Spot | HyperliquidProductType::Outcome)
        );
        let fetch_spot = filter_product != Some(HyperliquidProductType::Perp);

        let mut reports = Vec::new();
        let mut complete = true;
        let ts_init = self.clock.get_time_ns();

        if !fetch_perp {
            return self
                .request_spot_position_status_reports_sweep(user, instrument_id)
                .await;
        }

        for dex in dexes {
            let state_response = self
                .info_clearinghouse_state_for_dex(user, dex.as_deref())
                .await?;
            let asset_positions: Vec<serde_json::Value> = state_response
                .get("assetPositions")
                .and_then(|value| value.as_array())
                .ok_or_else(|| {
                    Error::bad_request("assetPositions not found in clearinghouse state")
                })?
                .clone();

            for position_value in asset_positions {
                let coin = position_value
                    .get("position")
                    .and_then(|position| position.get("coin"))
                    .and_then(|coin| coin.as_str())
                    .ok_or_else(|| Error::bad_request("coin not found in position"))?;

                let instrument = match self.get_or_create_instrument(&Ustr::from(coin), None) {
                    Some(instrument) => instrument,
                    // get_or_create_instrument warns with the coin
                    None => {
                        complete = false;
                        continue;
                    }
                };

                if instrument_id.is_some_and(|filter_id| instrument.id() != filter_id) {
                    continue;
                }

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

        // Spot positions are part of the report truth; propagate fetch errors
        // rather than silently omitting spot holdings from reconciliation.
        if fetch_spot {
            let spot_sweep = self
                .request_spot_position_status_reports_sweep(user, instrument_id)
                .await?;
            reports.extend(spot_sweep.reports);
            complete &= spot_sweep.complete;
        }

        Ok(ReportSweep { reports, complete })
    }

    /// Request account state (balances and margins) for a user.
    ///
    /// Fetches perp and spot clearinghouse state and the account abstraction mode from
    /// Hyperliquid and merges them into a single [`AccountState`]. For unified and portfolio
    /// margin accounts, balances come from the spot state alone and spot USDC `hold` is the
    /// account-wide margin. Otherwise USDC comes from the perp margin summary only when that
    /// summary reflects non-zero collateral, margin used, or withdrawable balance; if the
    /// summary is absent or zeroed, spot USDC is used instead. Non-USDC tokens are always
    /// appended from the spot balances. Spot tokens the venue lists at zero are reported at
    /// zero, and USDC is reported at zero when there is no USDC balance, so a previous balance
    /// is cleared. On accounts without spot collateral this needs a perp summary in the response.
    ///
    /// # Errors
    ///
    /// Returns an error if `account_id` is not set, or if the perp clearinghouse, spot
    /// clearinghouse, or user abstraction request fails. Spot and abstraction failures are
    /// propagated so the caller sees real API errors instead of a silently truncated or
    /// misread snapshot.
    pub async fn request_account_state(&self, user: &str) -> Result<AccountState> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let state_response = self.info_clearinghouse_state(user).await?;
        let ts_init = self.clock.get_time_ns();

        log::trace!("Clearinghouse state response: {state_response}");

        let perp_state: ClearinghouseState = serde_json::from_value(state_response.clone())
            .map_err(|e| {
                log::error!("Failed to parse clearinghouse state: {e}");
                log::debug!("Raw response: {state_response}");
                Error::bad_request(format!("Failed to parse clearinghouse state: {e}"))
            })?;

        // Spot must not be silently dropped: a 429 or parse error would
        // otherwise make non-USDC holdings look like they vanished.
        let spot_response = self.info_spot_clearinghouse_state(user).await?;
        let spot_state: SpotClearinghouseState = serde_json::from_value(spot_response.clone())
            .map_err(|e| {
                log::error!("Failed to parse spot clearinghouse state: {e}");
                log::debug!("Raw spot response: {spot_response}");
                Error::bad_request(format!("Failed to parse spot clearinghouse state: {e}"))
            })?;

        // Without the mode a unified account would be read from the default-dex perp summary.
        let abstraction = self.info_user_abstraction(user).await?;

        let (balances, margins) =
            parse_combined_account_balances_and_margins(&perp_state, &spot_state, abstraction)
                .map_err(|e| Error::decode(e.to_string()))?;

        Ok(AccountState::new(
            account_id,
            AccountType::Margin,
            balances,
            margins,
            true, // reported
            UUID4::new(),
            ts_init,
            ts_init,
            None,
        ))
    }

    /// Request spot token balances for a user.
    ///
    /// Fetches `spotClearinghouseState` and returns one [`AccountBalance`] per
    /// non-zero token. USDC is included as a separate balance entry when present;
    /// callers that also report perp margin state must dedupe currencies before
    /// emitting an [`AccountState`].
    ///
    /// # Errors
    ///
    /// Returns an error if the API request fails or the response cannot be parsed.
    pub async fn request_spot_balances(&self, user: &str) -> Result<Vec<AccountBalance>> {
        let response = self.info_spot_clearinghouse_state(user).await?;

        log::trace!("Spot clearinghouse state response: {response}");

        let state: SpotClearinghouseState =
            serde_json::from_value(response.clone()).map_err(|e| {
                log::error!("Failed to parse spot clearinghouse state: {e}");
                log::debug!("Raw response: {response}");
                Error::bad_request(format!("Failed to parse spot clearinghouse state: {e}"))
            })?;

        parse_spot_account_balances(&state).map_err(|e| Error::decode(e.to_string()))
    }

    /// Request spot position status reports for a user.
    ///
    /// Each non-zero spot balance is reported as a Long position against its
    /// `{BASE}-{QUOTE}-SPOT` instrument. HIP-4 outcome side tokens arrive on
    /// this same endpoint with `coin` set to the `+<encoding>` token form;
    /// those balances are resolved against the matching Outcome instrument so
    /// outcome holdings surface as positions through the standard reconcile
    /// path.
    ///
    /// # Errors
    ///
    /// Returns an error if `account_id` has not been set, the API request fails,
    /// or a non-zero balance cannot be resolved to an instrument or converted
    /// into a report (the snapshot is then incomplete and must not be treated
    /// as authoritative).
    pub async fn request_spot_position_status_reports(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<Vec<PositionStatusReport>> {
        let sweep = self
            .request_spot_position_status_reports_sweep(user, instrument_id)
            .await?;

        if !sweep.complete {
            return Err(Error::bad_request(
                "Spot position snapshot incomplete: at least one venue row could not be decoded, resolved to an instrument, or converted into a report",
            ));
        }

        Ok(sweep.reports)
    }

    pub(crate) async fn request_spot_position_status_reports_sweep(
        &self,
        user: &str,
        instrument_id: Option<InstrumentId>,
    ) -> Result<ReportSweep<PositionStatusReport>> {
        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let response = self.info_spot_clearinghouse_state(user).await?;

        let state: SpotClearinghouseState = serde_json::from_value(response).map_err(|e| {
            log::error!("Failed to parse spot clearinghouse state: {e}");
            Error::bad_request(format!("Failed to parse spot clearinghouse state: {e}"))
        })?;

        let ts_init = self.clock.get_time_ns();
        let mut reports = Vec::with_capacity(state.balances.len());
        let mut complete = true;

        for balance in &state.balances {
            if balance.total.is_zero() {
                continue;
            }

            // USDC is the universal quote for Hyperliquid spot: it funds every
            // pair and has no `USDC-*-SPOT` instrument. Skip it so the loop
            // does not trigger a misleading cache-miss WARN. Revisit if
            // Hyperliquid ever introduces a USDC-base spot pair.
            if balance.coin == "USDC" {
                continue;
            }

            let product_type = match HyperliquidProductType::from_symbol(balance.coin.as_str()) {
                Ok(HyperliquidProductType::Outcome) => HyperliquidProductType::Outcome,
                _ => HyperliquidProductType::Spot,
            };

            let instrument = match self.get_or_create_instrument(&balance.coin, Some(product_type))
            {
                Some(inst) => inst,
                // get_or_create_instrument warns with the coin
                None => {
                    complete = false;
                    continue;
                }
            };

            if let Some(filter_id) = instrument_id
                && instrument.id() != filter_id
            {
                continue;
            }

            match parse_spot_position_status_report(balance, &instrument, account_id, ts_init) {
                Ok(report) => reports.push(report),
                Err(e) => {
                    log::error!(
                        "Failed to parse spot position status report for {}: {e}",
                        balance.coin,
                    );
                    complete = false;
                }
            }
        }

        Ok(ReportSweep { reports, complete })
    }

    /// Request historical bars for an instrument.
    ///
    /// Fetches candle data from the Hyperliquid API and converts it to Nautilus bars.
    /// Incomplete bars (where end_timestamp >= current time) are filtered out.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The instrument is not found in cache.
    /// - The bar aggregation is unsupported by Hyperliquid.
    /// - The API request fails.
    /// - Parsing fails.
    ///
    /// # References
    ///
    /// <https://hyperliquid.gitbook.io/hyperliquid-docs/for-developers/api/info-endpoint#candles-snapshot>
    pub async fn request_bars(
        &self,
        bar_type: BarType,
        start: Option<jiff::Timestamp>,
        end: Option<jiff::Timestamp>,
        limit: Option<u32>,
    ) -> Result<Vec<Bar>> {
        let instrument_id = bar_type.instrument_id();
        let symbol = instrument_id.symbol;

        let product_type = HyperliquidProductType::from_symbol(symbol.as_str()).ok();

        // `cache_alias_for_symbol` mirrors how `cache_instrument` stores the
        // secondary key (token form `+<encoding>` for outcomes, leading
        // segment for perps / spots), so this lookup stays in sync.
        let alias = cache_alias_for_symbol(symbol.as_str())
            .map(|alias| Ustr::from(alias.as_str()))
            .ok_or_else(|| Error::bad_request("Invalid instrument symbol"))?;

        let instrument = self
            .get_or_create_instrument(&alias, product_type)
            .ok_or_else(|| {
                Error::bad_request(InstrumentLookupError::not_found(instrument_id).to_string())
            })?;

        // Use raw_symbol which has the correct Hyperliquid API format:
        // - Perps: base currency (e.g., "BTC")
        // - Spot PURR: slash format (e.g., "PURR/USDC")
        // - Spot others: @{index} format (e.g., "@107")
        let coin = instrument.raw_symbol().inner();

        let price_precision = instrument.price_precision();
        let size_precision = instrument.size_precision();

        let interval =
            bar_type_to_interval(&bar_type).map_err(|e| Error::bad_request(e.to_string()))?;

        // Hyperliquid uses millisecond timestamps
        let now = jiff::Timestamp::now();
        let end_time = end.unwrap_or(now).as_millisecond() as u64;
        let start_time = if let Some(start) = start {
            start.as_millisecond() as u64
        } else {
            // Default to 1000 bars before end_time
            let spec = bar_type.spec();
            let step_ms = match spec.aggregation {
                BarAggregation::Minute => spec.step.get() as u64 * 60_000,
                BarAggregation::Hour => spec.step.get() as u64 * 3_600_000,
                BarAggregation::Day => spec.step.get() as u64 * 86_400_000,
                BarAggregation::Week => spec.step.get() as u64 * 604_800_000,
                BarAggregation::Month => spec.step.get() as u64 * 2_592_000_000,
                _ => 60_000,
            };
            end_time.saturating_sub(1000 * step_ms)
        };

        let candles = self
            .info_candle_snapshot(coin.as_str(), interval, start_time, end_time)
            .await?;

        // Filter out incomplete bars where end_timestamp >= current time
        let now_ms = now.as_millisecond() as u64;

        let mut bars: Vec<Bar> = candles
            .iter()
            .filter(|candle| candle.end_timestamp < now_ms)
            .enumerate()
            .filter_map(|(i, candle)| {
                candle_to_bar(candle, bar_type, price_precision, size_precision)
                    .map_err(|e| {
                        log::error!("Failed to convert candle {i} to bar: {candle:?} error: {e}");
                        e
                    })
                    .ok()
            })
            .collect();

        // 0 means no limit
        if let Some(limit) = limit
            && limit > 0
            && bars.len() > limit as usize
        {
            bars.truncate(limit as usize);
        }

        log::debug!(
            "Received {} bars for {} (filtered {} incomplete)",
            bars.len(),
            bar_type,
            candles.len() - bars.len()
        );
        Ok(bars)
    }

    /// Request the recent public trade snapshot for an instrument.
    ///
    /// Hyperliquid's `recentTrades` endpoint is a bounded newest-first snapshot,
    /// rather than a range-query endpoint. The returned trades are normalized to
    /// ascending event time and then constrained to the requested window.
    ///
    /// A self-hosted node without the indexer responds with HTTP 422. This is
    /// treated as no available coverage so requests can still complete.
    pub async fn request_public_trades(
        &self,
        instrument_id: InstrumentId,
        start: Option<jiff::Timestamp>,
        end: Option<jiff::Timestamp>,
        limit: Option<usize>,
    ) -> Result<Vec<HyperliquidPublicTrade>> {
        let symbol = instrument_id.symbol;
        let product_type = HyperliquidProductType::from_symbol(symbol.as_str()).ok();
        let alias = cache_alias_for_symbol(symbol.as_str())
            .map(|alias| Ustr::from(alias.as_str()))
            .ok_or_else(|| Error::bad_request("Invalid instrument symbol"))?;
        let instrument = self
            .get_or_create_instrument(&alias, product_type)
            .ok_or_else(|| {
                Error::bad_request(InstrumentLookupError::not_found(instrument_id).to_string())
            })?;

        let raw_trades = match self
            .info_recent_trades(instrument.raw_symbol().as_ref())
            .await
        {
            Ok(trades) => trades,
            Err(e) if e.is_unprocessable_entity() => {
                log::warn!(
                    "Recent public trades endpoint unavailable for {instrument_id} \
                     (requires the Hyperliquid indexer); returning empty response"
                );
                Vec::new()
            }
            Err(e) => return Err(e),
        };

        let mut trades: Vec<HyperliquidPublicTrade> = raw_trades
            .iter()
            .filter_map(|raw| match parse_recent_public_trade(raw, &instrument) {
                Ok(trade) => Some(trade),
                Err(e) => {
                    log::warn!("Skipping recent public trade for {instrument_id}: {e}");
                    None
                }
            })
            .collect();
        trades.sort_by_key(|trade| trade.ts_event);

        Ok(filter_recent_public_trades(
            trades,
            datetime_to_unix_nanos(start),
            datetime_to_unix_nanos(end),
            limit.filter(|limit| *limit > 0),
            instrument_id,
        ))
    }

    /// Requests funding rate history for a perpetual, paging past the venue's
    /// per-response cap.
    ///
    /// With `start`, returns up to `limit` rows from `start` onward. Without it,
    /// returns the most recent `limit` rows (default seven days).
    ///
    /// # Errors
    ///
    /// Returns an error if the instrument is not a perpetual, `start` is not
    /// before `end`, or a request fails.
    pub async fn request_funding_rates(
        &self,
        instrument: &InstrumentAny,
        start: Option<jiff::Timestamp>,
        end: Option<jiff::Timestamp>,
        limit: Option<usize>,
    ) -> Result<Vec<FundingRateUpdate>> {
        let instrument_id = instrument.id();

        if !matches!(instrument, InstrumentAny::CryptoPerpetual(_)) {
            return Err(Error::bad_request(format!(
                "funding rates are only available for perpetual instruments: {instrument_id}",
            )));
        }

        let coin = instrument.raw_symbol().inner();

        if let (Some(start), Some(end)) = (start, end)
            && start >= end
        {
            return Err(Error::bad_request(format!(
                "invalid funding request range: start={start}, end={end}",
            )));
        }

        let now = jiff::Timestamp::now();
        let end = end.unwrap_or(now).min(now);

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

        let requested_limit = limit.filter(|n| *n > 0);
        let target_limit = requested_limit.unwrap_or(DEFAULT_FUNDING_RATES_LIMIT);
        let start_was_unspecified = start.is_none();
        let end_ms = end.as_millisecond().max(0) as u64;

        let start_ms = start.map_or_else(
            || {
                let lookback_rows = target_limit.saturating_add(1) as u64;
                // Hyperliquid funding interval is 1h
                end_ms.saturating_sub((60 * 60 * 1_000_u64).saturating_mul(lookback_rows))
            },
            |dt| dt.as_millisecond().max(0) as u64,
        );

        let mut funding_rates = Vec::new();
        let mut cursor_ms = start_ms;

        'pages: while cursor_ms <= end_ms {
            let entries = self
                .info_funding_history(coin.as_str(), cursor_ms, Some(end_ms))
                .await?;
            let Some(last_time_ms) = entries.iter().map(|entry| entry.time).max() else {
                break;
            };

            let mut page: Vec<FundingRateUpdate> = entries
                .iter()
                .filter(|entry| (cursor_ms..=end_ms).contains(&entry.time))
                .map(|entry| funding_entry_to_update(entry, instrument_id))
                .collect();
            page.sort_by_key(|rate| rate.ts_event);

            for update in page {
                if funding_rates
                    .last()
                    .is_some_and(|last: &FundingRateUpdate| last.ts_event == update.ts_event)
                {
                    continue;
                }
                funding_rates.push(update);

                if !start_was_unspecified
                    && requested_limit.is_some_and(|limit| funding_rates.len() >= limit)
                {
                    break 'pages;
                }
            }

            if entries.len() < HYPERLIQUID_FUNDING_HISTORY_MAX_LIMIT {
                break;
            }

            // `startTime` is inclusive, so step past the last row
            let next_cursor_ms = last_time_ms.saturating_add(1);
            if next_cursor_ms <= cursor_ms {
                break;
            }
            cursor_ms = next_cursor_ms;
        }

        if start_was_unspecified && funding_rates.len() > target_limit {
            funding_rates = funding_rates.split_off(funding_rates.len() - target_limit);
        }

        Ok(funding_rates)
    }

    /// Submits an order to the exchange.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, order validation fails, serialization fails,
    /// or the API returns an error.
    #[expect(clippy::too_many_arguments)]
    pub async fn submit_order(
        &self,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: TimeInForce,
        price: Option<Price>,
        trigger_price: Option<Price>,
        post_only: bool,
        reduce_only: bool,
    ) -> Result<OrderStatusReport> {
        let symbol = instrument_id.symbol.inner();
        let asset = self.get_asset_index_for_symbol(symbol).ok_or_else(|| {
            Error::bad_request(format!(
                "Asset index not found for symbol: {symbol}. Ensure instruments are loaded."
            ))
        })?;

        let is_buy = matches!(order_side, OrderSide::Buy);
        let price_precision = self.get_price_precision_for_symbol(symbol);

        let price_decimal = match price {
            Some(px) => normalize_or_validate_wire_price(
                px.as_decimal(),
                "Price",
                price_precision,
                self.normalize_prices,
            )
            .map_err(|e| Error::bad_request(format!("{e}")))?,
            None if matches!(order_type, OrderType::Market) => Decimal::ZERO,
            None if matches!(
                order_type,
                OrderType::StopMarket | OrderType::MarketIfTouched
            ) =>
            {
                match trigger_price {
                    Some(tp) => {
                        let derived = derive_limit_from_trigger(
                            tp.as_decimal().normalize(),
                            is_buy,
                            self.market_order_slippage_bps,
                        );
                        let sig_rounded = round_to_sig_figs(derived, 5);
                        clamp_price_to_precision(sig_rounded, price_precision.unwrap_or(2), is_buy)
                            .normalize()
                    }
                    None => Decimal::ZERO,
                }
            }
            None => return Err(Error::bad_request("Limit orders require a price")),
        };

        let size_decimal = quantity.as_decimal().normalize();

        let kind = match order_type {
            OrderType::Market => HyperliquidExchangeOrderKind::Limit {
                limit: HyperliquidExchangeLimitParams {
                    tif: HyperliquidExchangeTif::Ioc,
                },
            },
            OrderType::Limit => {
                let tif = if post_only {
                    HyperliquidExchangeTif::Alo
                } else {
                    match time_in_force {
                        TimeInForce::Gtc => HyperliquidExchangeTif::Gtc,
                        TimeInForce::Ioc => HyperliquidExchangeTif::Ioc,
                        TimeInForce::Fok
                        | TimeInForce::Day
                        | TimeInForce::Gtd
                        | TimeInForce::AtTheOpen
                        | TimeInForce::AtTheClose => {
                            return Err(Error::bad_request(format!(
                                "Time in force {time_in_force:?} not supported"
                            )));
                        }
                    }
                };
                HyperliquidExchangeOrderKind::Limit {
                    limit: HyperliquidExchangeLimitParams { tif },
                }
            }
            OrderType::StopMarket
            | OrderType::StopLimit
            | OrderType::MarketIfTouched
            | OrderType::LimitIfTouched => {
                if let Some(trig_px) = trigger_price {
                    let trigger_price_decimal = normalize_or_validate_wire_price(
                        trig_px.as_decimal(),
                        "Trigger price",
                        price_precision,
                        self.normalize_prices,
                    )
                    .map_err(|e| Error::bad_request(format!("{e}")))?;

                    // Determine TP/SL type based on order type
                    // StopMarket/StopLimit are always Sl (protective stops)
                    // MarketIfTouched/LimitIfTouched are always Tp (profit-taking/entry)
                    let tpsl = match order_type {
                        OrderType::StopMarket | OrderType::StopLimit => HyperliquidExchangeTpSl::Sl,
                        OrderType::MarketIfTouched | OrderType::LimitIfTouched => {
                            HyperliquidExchangeTpSl::Tp
                        }
                        _ => unreachable!(),
                    };

                    let is_market = matches!(
                        order_type,
                        OrderType::StopMarket | OrderType::MarketIfTouched
                    );

                    HyperliquidExchangeOrderKind::Trigger {
                        trigger: HyperliquidExchangeTriggerParams {
                            is_market,
                            trigger_px: trigger_price_decimal,
                            tpsl,
                        },
                    }
                } else {
                    return Err(Error::bad_request("Trigger orders require a trigger price"));
                }
            }
            _ => {
                return Err(Error::bad_request(format!(
                    "Order type {order_type:?} not supported"
                )));
            }
        };

        let cloid = self.get_or_generate_client_order_id_cloid(client_order_id);
        let hyperliquid_order = HyperliquidExchangePlaceOrderRequest {
            asset,
            is_buy,
            price: price_decimal,
            size: size_decimal,
            reduce_only,
            kind,
            cloid: Some(cloid),
        };

        let builder = self.builder_attribution();

        let action = HyperliquidExchangeAction::Order {
            orders: vec![hyperliquid_order],
            grouping: HyperliquidExchangeGrouping::Na,
            builder,
        };

        let response = self.inner.post_action_exec(&action).await?;

        // A single (non-bracket) order should return an actionable status;
        // `None` (a deferred `Tag` child) is unexpected on this HTTP path.
        self.build_submit_order_report(
            instrument_id,
            client_order_id,
            order_side,
            order_type,
            quantity,
            time_in_force,
            price,
            trigger_price,
            response,
        )?
        .ok_or_else(|| {
            Error::bad_request(
                "Single-order submission returned no actionable status (deferred trigger child)",
            )
        })
    }

    /// Submit an order using an OrderAny object.
    ///
    /// This is a convenience method that wraps submit_order.
    ///
    /// # Errors
    ///
    /// Returns an error if the quantity is quote-denominated: this raw path has no cached market
    /// data for a quote-to-base conversion, so the order must be submitted through the execution
    /// client instead.
    pub async fn submit_order_from_order_any(&self, order: &OrderAny) -> Result<OrderStatusReport> {
        if order.is_quote_quantity() {
            return Err(Error::bad_request(
                "Quote-denominated quantity orders must submit through the execution client \
                 for quote-to-base conversion",
            ));
        }

        self.submit_order(
            order.instrument_id(),
            order.client_order_id(),
            order.order_side(),
            order.order_type(),
            order.quantity(),
            order.time_in_force(),
            order.price(),
            order.trigger_price(),
            order.is_post_only(),
            order.is_reduce_only(),
        )
        .await
    }

    #[expect(clippy::too_many_arguments)]
    fn create_order_status_report(
        &self,
        instrument_id: InstrumentId,
        client_order_id: Option<ClientOrderId>,
        venue_order_id: VenueOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: TimeInForce,
        price: Option<Price>,
        trigger_price: Option<Price>,
        order_status: OrderStatus,
        filled_qty: Quantity,
        _instrument: &InstrumentAny,
        account_id: AccountId,
        ts_init: UnixNanos,
    ) -> OrderStatusReport {
        let ts_accepted = self.clock.get_time_ns();
        let ts_last = ts_accepted;
        let report_id = UUID4::new();

        let mut report = OrderStatusReport::new(
            account_id,
            instrument_id,
            client_order_id,
            venue_order_id,
            order_side.into(),
            order_type,
            time_in_force,
            order_status,
            quantity,
            filled_qty,
            ts_accepted,
            ts_last,
            ts_init,
            Some(report_id),
        );

        if let Some(px) = price {
            report = report.with_price(px);
        }

        if let Some(trig_px) = trigger_price {
            report = report
                .with_trigger_price(trig_px)
                .with_trigger_type(TriggerType::Default);
        }

        report
    }

    /// Submit multiple orders to the Hyperliquid exchange in a single request.
    ///
    /// # Errors
    ///
    /// Returns an error if credentials are missing, order validation fails, serialization fails,
    /// or the API returns an error. Also returns an error for any quote-denominated quantity:
    /// this raw path has no cached market data for a quote-to-base conversion, so such orders
    /// must be submitted through the execution client instead.
    pub async fn submit_orders(&self, orders: &[&OrderAny]) -> Result<Vec<OrderStatusReport>> {
        // Convert orders using asset indices from the cached map
        let mut hyperliquid_orders = Vec::with_capacity(orders.len());
        let mut client_order_ids = Vec::with_capacity(orders.len());

        for order in orders {
            if order.is_quote_quantity() {
                return Err(Error::bad_request(format!(
                    "Quote-denominated quantity order {} must submit through the execution \
                     client for quote-to-base conversion",
                    order.client_order_id()
                )));
            }

            let instrument_id = order.instrument_id();
            let symbol = instrument_id.symbol.inner();
            let asset = self.get_asset_index_for_symbol(symbol).ok_or_else(|| {
                Error::bad_request(format!(
                    "Asset index not found for symbol: {symbol}. Ensure instruments are loaded."
                ))
            })?;
            let price_decimals = self.get_price_precision_for_symbol(symbol);
            let request = order_to_hyperliquid_request_with_optional_decimals(
                order,
                asset,
                price_decimals,
                self.normalize_prices,
                self.market_order_slippage_bps,
                None,
            )
            .map_err(|e| Error::bad_request(format!("Failed to convert order: {e}")))?;
            client_order_ids.push(order.client_order_id());
            hyperliquid_orders.push(request);
        }

        for (request, client_order_id) in hyperliquid_orders.iter_mut().zip(client_order_ids) {
            request.cloid = Some(self.get_or_generate_client_order_id_cloid(client_order_id));
        }

        let builder = self.builder_attribution();

        let grouping =
            determine_order_list_grouping(&orders.iter().copied().cloned().collect::<Vec<_>>());

        let action = HyperliquidExchangeAction::Order {
            orders: hyperliquid_orders,
            grouping,
            builder,
        };

        // Submit to exchange using the typed exec endpoint
        let response = self.inner.post_action_exec(&action).await?;

        self.build_submit_orders_reports(orders, grouping, response)
    }

    /// Parses a Hyperliquid exchange order response for a single-order submit
    /// into an [`OrderStatusReport`].
    ///
    /// Returns `Ok(None)` when the venue returned an empty `statuses` array or
    /// when the only status is a deferred `Tag` child (for example
    /// `waitingForFill`): the venue accepted the order but has not assigned an
    /// oid yet, so the order stays `SUBMITTED` until the user-events stream
    /// drives the first `OrderAccepted` with the real oid.
    ///
    /// Shared by the HTTP and WebSocket single-submit paths.
    ///
    /// # Errors
    ///
    /// Returns an error if account credentials are missing, the response is
    /// malformed, or the order returned an `error` status.
    #[expect(clippy::too_many_arguments)]
    pub fn build_submit_order_report(
        &self,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: TimeInForce,
        price: Option<Price>,
        trigger_price: Option<Price>,
        response: HyperliquidExchangeResponse,
    ) -> Result<Option<OrderStatusReport>> {
        let order_response = parse_order_response(response)?;

        let Some(order_status) = order_response.statuses.first() else {
            return Ok(None);
        };

        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let ts_init = self.clock.get_time_ns();

        self.build_status_report(
            instrument_id,
            client_order_id,
            order_side,
            order_type,
            quantity,
            time_in_force,
            price,
            trigger_price,
            order_status,
            account_id,
            ts_init,
        )
    }

    /// Parses a Hyperliquid exchange order response into per-order
    /// [`OrderStatusReport`]s, paired positionally with `orders`.
    ///
    /// Shared by the HTTP and WebSocket batch-submit paths since the response
    /// envelope is identical regardless of transport. Deferred `Tag` children
    /// (for example `waitingForFill`) are elided from the result; those orders
    /// stay `SUBMITTED` until the user-events stream delivers an `OrderAccepted`
    /// with the real oid.
    ///
    /// # Errors
    ///
    /// Returns an error if account credentials are missing, the response is
    /// malformed, an order returned an `error` status, or, for ungrouped
    /// submissions, the response status count diverges from the order count.
    pub fn build_submit_orders_reports(
        &self,
        orders: &[&OrderAny],
        grouping: HyperliquidExchangeGrouping,
        response: HyperliquidExchangeResponse,
    ) -> Result<Vec<OrderStatusReport>> {
        let order_response = parse_order_response(response)?;

        let account_id = self
            .account_id
            .ok_or_else(|| Error::bad_request("Account ID not set"))?;
        let ts_init = self.clock.get_time_ns();

        // For grouped orders (NormalTpsl/PositionTpsl) the exchange returns a
        // single status for the whole group, so only enforce 1:1 matching for
        // ungrouped (Na) submissions.
        if grouping == HyperliquidExchangeGrouping::Na
            && order_response.statuses.len() != orders.len()
        {
            return Err(Error::bad_request(format!(
                "Mismatch between submitted orders ({}) and response statuses ({})",
                orders.len(),
                order_response.statuses.len()
            )));
        }

        // The exchange returns statuses in submission order, so pair each order
        // with its status positionally.
        let mut reports = Vec::with_capacity(order_response.statuses.len());
        for (order, order_status) in orders.iter().zip(order_response.statuses.iter()) {
            if let Some(report) = self.build_status_report(
                order.instrument_id(),
                order.client_order_id(),
                order.order_side(),
                order.order_type(),
                order.quantity(),
                order.time_in_force(),
                order.price(),
                order.trigger_price(),
                order_status,
                account_id,
                ts_init,
            )? {
                reports.push(report);
            }
        }

        Ok(reports)
    }

    /// Builds an [`OrderStatusReport`] from a single venue status, or `Ok(None)`
    /// for a deferred `Tag` child that has no oid yet.
    ///
    /// `Tag` rows are elided rather than given a synthetic placeholder venue id:
    /// an earlier placeholder accept was deduped against the later real accept,
    /// so the cache never picked up the real oid and cancel/modify by venue id
    /// broke on bracket children.
    #[expect(clippy::too_many_arguments)]
    fn build_status_report(
        &self,
        instrument_id: InstrumentId,
        client_order_id: ClientOrderId,
        order_side: OrderSide,
        order_type: OrderType,
        quantity: Quantity,
        time_in_force: TimeInForce,
        price: Option<Price>,
        trigger_price: Option<Price>,
        order_status: &HyperliquidExchangeOrderStatus,
        account_id: AccountId,
        ts_init: UnixNanos,
    ) -> Result<Option<OrderStatusReport>> {
        if matches!(order_status, HyperliquidExchangeOrderStatus::Tag(_)) {
            return Ok(None);
        }

        let symbol = instrument_id.symbol.as_str();
        let product_type = HyperliquidProductType::from_symbol(symbol).ok();

        // Mirror the alias `cache_instrument` stored (token form for outcomes,
        // leading segment for perps / spots).
        let asset = cache_alias_for_symbol(symbol).unwrap_or_else(|| symbol.to_string());
        let instrument = self
            .get_or_create_instrument(&Ustr::from(asset.as_str()), product_type)
            .ok_or_else(|| {
                Error::bad_request(InstrumentLookupError::not_found(instrument_id).to_string())
            })?;

        let report = match order_status {
            HyperliquidExchangeOrderStatus::Resting { resting } => self.create_order_status_report(
                instrument_id,
                Some(client_order_id),
                VenueOrderId::new(resting.oid.to_string()),
                order_side,
                order_type,
                quantity,
                time_in_force,
                price,
                trigger_price,
                OrderStatus::Accepted,
                Quantity::zero(instrument.size_precision()),
                &instrument,
                account_id,
                ts_init,
            ),
            HyperliquidExchangeOrderStatus::Filled { filled } => {
                let filled_qty =
                    Quantity::from_decimal_dp(filled.total_sz, instrument.size_precision())
                        .map_err(|e| {
                            Error::bad_request(format!(
                                "Invalid filled size {}: {e}",
                                filled.total_sz
                            ))
                        })?;
                self.create_order_status_report(
                    instrument_id,
                    Some(client_order_id),
                    VenueOrderId::new(filled.oid.to_string()),
                    order_side,
                    order_type,
                    quantity,
                    time_in_force,
                    price,
                    trigger_price,
                    OrderStatus::Filled,
                    filled_qty,
                    &instrument,
                    account_id,
                    ts_init,
                )
            }
            HyperliquidExchangeOrderStatus::Error { error } => {
                return Err(Error::bad_request(format!(
                    "Order {client_order_id} rejected: {error}"
                )));
            }
            HyperliquidExchangeOrderStatus::Tag(_) => unreachable!("handled above"),
        };

        Ok(Some(report))
    }

    fn reconciliation_dexes(&self, instrument_id: Option<InstrumentId>) -> Vec<Option<Ustr>> {
        if let Some(instrument_id) = instrument_id {
            return vec![perp_dex_from_symbol(instrument_id.symbol.as_str())];
        }

        let cached = self.instruments.load();
        reconciliation_dexes_from_builders(
            cached
                .keys()
                .filter_map(|symbol| perp_dex_from_symbol(symbol.as_str())),
        )
    }

    pub(crate) async fn reconciliation_dexes_from_activity(
        &self,
        historical_orders: &[HyperliquidOrderStatusEntry],
        fills: &HyperliquidFills,
    ) -> Result<Vec<Option<Ustr>>> {
        if historical_orders.len() >= HYPERLIQUID_RECENT_HISTORY_LIMIT
            || fills.len() >= HYPERLIQUID_RECENT_HISTORY_LIMIT
        {
            let builder_dexes = self
                .inner
                .load_perp_dexs()
                .await?
                .into_iter()
                .flatten()
                .filter(|dex| !dex.name.is_empty())
                .map(|dex| Ustr::from(dex.name.as_str()));
            return Ok(reconciliation_dexes_from_builders(builder_dexes));
        }

        let builder_dexes = historical_orders
            .iter()
            .map(|entry| entry.order.coin.as_str())
            .chain(fills.iter().map(|fill| fill.coin.as_str()))
            .filter_map(perp_dex_from_activity_coin);

        Ok(reconciliation_dexes_from_builders(builder_dexes))
    }
}

// A reconciliation snapshot with its completeness flag: `complete` is false when
// at least one venue row could not be decoded, resolved to an instrument, or
// converted into a report. Callers whose contract cannot carry the flag fail
// closed; mass-status reconciliation preserves the valid rows and reports the
// incompleteness via `ExecutionMassStatus::set_report_window`.
#[derive(Debug)]
pub(crate) struct ReportSweep<T> {
    pub reports: Vec<T>,
    pub complete: bool,
}

/// The instrument universe with the market status of each instrument.
#[derive(Debug)]
pub(crate) struct InstrumentSnapshot {
    pub instruments: Vec<InstrumentAny>,
    pub statuses: AHashMap<InstrumentId, MarketStatusAction>,
    /// Product types whose metadata loaded in full.
    pub loaded: AHashSet<HyperliquidProductType>,
}

fn reconciliation_dexes_from_builders(
    builder_dexes: impl IntoIterator<Item = Ustr>,
) -> Vec<Option<Ustr>> {
    let mut builder_dexes = builder_dexes.into_iter().collect::<Vec<_>>();
    builder_dexes.sort_unstable();
    builder_dexes.dedup();

    let mut dexes = Vec::with_capacity(builder_dexes.len() + 1);
    dexes.push(None);
    dexes.extend(builder_dexes.into_iter().map(Some));
    dexes
}

fn perp_dex_from_activity_coin(coin: &str) -> Option<Ustr> {
    if coin.starts_with(VAULT_TOKEN_PREFIX) {
        return None;
    }

    let (dex, _) = coin.split_once(':')?;
    (!dex.is_empty()).then(|| Ustr::from(dex))
}

fn perp_dex_from_symbol(symbol: &str) -> Option<Ustr> {
    symbol
        .strip_suffix("-PERP")?
        .split_once(':')
        .map(|(dex, _)| Ustr::from(dex))
}

/// Extracts the order-status payload from an exchange response.
///
/// The newer response format nests the statuses under `data`; the older format
/// places them directly in the response body.
fn parse_order_response(
    response: HyperliquidExchangeResponse,
) -> Result<HyperliquidExchangeOrderResponseData> {
    let response_data = match response {
        HyperliquidExchangeResponse::Status {
            status,
            response: response_data,
        } if status == RESPONSE_STATUS_OK => response_data,
        HyperliquidExchangeResponse::Error { error } => {
            return Err(Error::bad_request(format!(
                "Order submission failed: {error}"
            )));
        }
        _ => return Err(Error::bad_request("Unexpected response format")),
    };

    let data_value = if let Some(data) = response_data.get("data") {
        data.clone()
    } else {
        response_data
    };

    serde_json::from_value(data_value)
        .map_err(|e| Error::bad_request(format!("Failed to parse order response: {e}")))
}

fn resolve_perp_dex_name(
    dex_index: usize,
    meta: &PerpMeta,
    perp_dexs: Option<&[Option<PerpDex>]>,
) -> String {
    if dex_index == 0 {
        return String::new();
    }

    if let Some(dex_name) = perp_dexs
        .and_then(|dexs| dexs.get(dex_index))
        .and_then(|dex| dex.as_ref())
        .map(|dex| dex.name.clone())
    {
        return dex_name;
    }

    meta.universe
        .iter()
        .find_map(|asset| asset.name.split_once(':').map(|(dex, _)| dex.to_string()))
        .unwrap_or_default()
}

/// Returns the asset index base for a perp dex.
///
/// Standard perps (dex 0) start at 0. HIP-3 dexes start at
/// 100_000 + dex_index * 10_000.
fn perp_dex_asset_index_base(dex_index: usize) -> u32 {
    if dex_index == 0 {
        0
    } else {
        100_000 + dex_index as u32 * 10_000
    }
}

fn cache_coin_aliases(
    cache: &mut AHashMap<(Ustr, HyperliquidProductType), InstrumentAny>,
    instrument: &InstrumentAny,
    product_type: HyperliquidProductType,
) {
    let full_symbol = instrument.symbol().inner();
    let coin = instrument.raw_symbol().inner();

    // Composite key allows disambiguating same coin across PERP and SPOT
    cache.insert((coin, product_type), instrument.clone());

    // Secondary alias key for two distinct callers:
    //
    // * Spot raw_symbols are either `@{pair_index}` or slash format
    //   (e.g., "PURR/USDC"); spot balance/position reconciliation
    //   maps the venue token name (e.g., "PURR") to instruments via
    //   this alias.
    // * Order submission paths split `instrument_id.symbol` on `-`
    //   to derive a coin key. For HIP-3 perps with wildcard-bearing
    //   venue names, the sanitized base in `instrument_id.symbol`
    //   (e.g., "dex:STREAMABCDxxxx") differs from `raw_symbol` /
    //   `coin` (e.g., "dex:STREAMABCD****"), so an alias on the
    //   sanitized base lets that lookup resolve.
    //
    // For outcomes the alias is the `+<encoding>` token form
    // (matching the `coin` field on `spotClearinghouseState`);
    // for perps / spots it is the leading symbol segment.
    // `cache_alias_for_symbol` keeps the two rules co-located so
    // every caller derives the same key.
    //
    // Keep aliases on their original venue market while refreshing
    // its definition, canonical spot pairs are cached first.
    if let Some(alias_ustr) =
        cache_alias_for_symbol(full_symbol.as_str()).map(|alias| Ustr::from(alias.as_str()))
    {
        let key = (alias_ustr, product_type);
        if alias_ustr != coin
            && cache
                .get(&key)
                .is_none_or(|cached| cached.raw_symbol() == instrument.raw_symbol())
        {
            cache.insert(key, instrument.clone());
        }
    }
}

fn validate_schedule_cancel_time(time: u64, now: u64) -> Result<()> {
    if time.checked_sub(now).is_none_or(|delay| delay < 5_000) {
        return Err(Error::bad_request(
            "Schedule cancel time must be at least 5000 ms after the current time",
        ));
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{
        collections::HashMap,
        net::SocketAddr,
        sync::{
            Arc,
            atomic::{AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use axum::{
        Router,
        extract::State,
        http::StatusCode,
        response::{IntoResponse, Json, Response},
        routing::post,
    };
    use nautilus_core::{Params, UUID4, UnixNanos, hex, time::get_atomic_clock_realtime};
    use nautilus_model::{
        currencies::CURRENCY_MAP,
        enums::{CurrencyType, OrderSide, OrderStatus, OrderType, TimeInForce, TriggerType},
        events::{OrderEventAny, OrderInitialized},
        identifiers::{
            AccountId, ClientOrderId, InstrumentId, StrategyId, Symbol, TraderId, VenueOrderId,
        },
        instruments::{CryptoPerpetual, CurrencyPair, Instrument, InstrumentAny},
        orders::OrderAny,
        reports::OrderStatusReport,
        types::{Currency, Price, Quantity},
    };
    use nautilus_testkit::http::assert_http_redirect_rejected;
    use rstest::rstest;
    use rust_decimal_macros::dec;
    use serde_json::{Value, json};
    use ustr::Ustr;

    use super::{
        HyperliquidHttpClient, HyperliquidRawHttpClient, RETRY_AFTER_HEADER, resolve_perp_dex_name,
        validate_schedule_cancel_time,
    };
    use crate::{
        common::{
            consts::{ASSET_INDEX_INFO_KEY, HYPERLIQUID_VENUE, NAUTILUS_BUILDER_ADDRESS},
            credential::EvmPrivateKey,
            enums::{HyperliquidEnvironment, HyperliquidProductType},
        },
        http::{
            error::Error,
            models::{
                Cloid, HyperliquidExchangeAction, HyperliquidExchangeResponse,
                HyperliquidOrderStatusEntry, PerpAsset, PerpDex, PerpMeta,
            },
            query::InfoRequest,
        },
        signing::{HyperliquidActionType, HyperliquidEip712Signer, SignRequest, nonce::TimeNonce},
    };

    const TEST_PRIVATE_KEY: &str =
        "0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef";

    #[rstest]
    #[case(10_000, 9_999, false)]
    #[case(10_000, 10_000, false)]
    #[case(10_000, 14_999, false)]
    #[case(10_000, 15_000, true)]
    #[case(10_000, 15_001, true)]
    #[case(u64::MAX - 4_999, u64::MAX, false)]
    #[case(u64::MAX - 5_000, u64::MAX, true)]
    fn test_schedule_cancel_timestamp(#[case] now: u64, #[case] time: u64, #[case] valid: bool) {
        assert_eq!(validate_schedule_cancel_time(time, now).is_ok(), valid);
    }

    #[rstest]
    fn test_usd_class_transfer_rejected_without_credentials(
        #[values(HyperliquidEnvironment::Mainnet, HyperliquidEnvironment::Testnet)]
        environment: HyperliquidEnvironment,
    ) {
        let client = HyperliquidRawHttpClient::new(environment, 5, None).unwrap();

        let action = HyperliquidExchangeAction::UsdClassTransfer {
            from: "spot".to_string(),
            to: "perp".to_string(),
            amount: dec!(1.25),
        };

        let error = client.sign_action_exec_request(&action, None).unwrap_err();

        assert!(matches!(error, Error::BadRequest(_)));
        assert_eq!(
            error.to_string(),
            "bad request: Unsupported action: usdClassTransfer"
        );
    }

    #[rstest]
    fn test_usd_class_transfer_rejected_before_nonce_allocation(
        #[values(HyperliquidEnvironment::Mainnet, HyperliquidEnvironment::Testnet)]
        environment: HyperliquidEnvironment,
        #[values(None, Some("0x1234567890123456789012345678901234567890"))] vault: Option<&str>,
        #[values(None, Some(1_900_000_000_000))] expires_after: Option<u64>,
    ) {
        let mut client =
            HyperliquidHttpClient::from_credentials(TEST_PRIVATE_KEY, vault, environment, 5, None)
                .unwrap();
        Arc::get_mut(&mut client.inner).unwrap().nonce_manager = None;

        let action = HyperliquidExchangeAction::UsdClassTransfer {
            from: "perp".to_string(),
            to: "spot".to_string(),
            amount: dec!(2.75),
        };

        let error = client
            .sign_action_exec_request(&action, expires_after)
            .unwrap_err();

        assert!(matches!(error, Error::BadRequest(_)));
        assert_eq!(
            error.to_string(),
            "bad request: Unsupported action: usdClassTransfer"
        );
    }

    #[rstest]
    #[tokio::test]
    async fn test_usd_class_transfer_rejected_without_dispatch(
        #[values(HyperliquidEnvironment::Mainnet, HyperliquidEnvironment::Testnet)]
        environment: HyperliquidEnvironment,
    ) {
        let (addr, count, server) =
            start_retry_server(RetryFailure::Response(400), usize::MAX, None).await;
        let mut client =
            HyperliquidHttpClient::from_credentials(TEST_PRIVATE_KEY, None, environment, 5, None)
                .unwrap();
        client.set_base_exchange_url(format!("http://{addr}/exchange"));

        let action = HyperliquidExchangeAction::UsdClassTransfer {
            from: "spot".to_string(),
            to: "perp".to_string(),
            amount: dec!(3.125),
        };

        let error = client.post_action_exec(&action).await.unwrap_err();

        server.abort();

        assert!(matches!(error, Error::BadRequest(_)));
        assert_eq!(
            error.to_string(),
            "bad request: Unsupported action: usdClassTransfer"
        );
        assert_eq!(count.load(Ordering::Relaxed), 0);
    }

    #[rstest]
    #[case(0)]
    #[case(1)]
    #[case(2)]
    #[case(3)]
    fn test_leverage_margin_signed_request(
        #[case] case_index: usize,
        #[values(HyperliquidEnvironment::Mainnet, HyperliquidEnvironment::Testnet)]
        environment: HyperliquidEnvironment,
    ) {
        let fixture: Value =
            serde_json::from_str(include_str!("../../test_data/leverage_margin_actions.json"))
                .unwrap();
        let case = &fixture["cases"][case_index];
        let action: HyperliquidExchangeAction =
            serde_json::from_str(case["json"].as_str().unwrap()).unwrap();
        let client =
            HyperliquidHttpClient::from_credentials(TEST_PRIVATE_KEY, None, environment, 60, None)
                .unwrap();
        let expires_after = Some(1_900_000_000_000);
        let request = client
            .sign_action_exec_request(&action, expires_after)
            .unwrap();
        let key = EvmPrivateKey::new(TEST_PRIVATE_KEY).unwrap();
        let signer = HyperliquidEip712Signer::new(&key).unwrap();
        let expected_bytes = hex::decode(case["msgpack"].as_str().unwrap()).unwrap();

        let expected_signature = signer
            .sign(&SignRequest {
                action: None,
                action_bytes: Some(expected_bytes),
                time_nonce: TimeNonce::from_millis(i128::from(request.nonce)),
                action_type: HyperliquidActionType::L1,
                is_testnet: environment == HyperliquidEnvironment::Testnet,
                vault_address: None,
                expires_after,
            })
            .unwrap()
            .signature;

        let expected_action: Value = serde_json::from_str(case["json"].as_str().unwrap()).unwrap();

        assert_eq!(
            serde_json::to_value(&request).unwrap(),
            json!({
                "action": expected_action,
                "nonce": request.nonce,
                "signature": expected_signature,
                "expiresAfter": expires_after,
            })
        );
    }

    #[tokio::test]
    async fn test_authenticated_client_rejects_redirects() {
        let client = HyperliquidRawHttpClient::build_http_client(3, None).unwrap();
        assert_http_redirect_rejected(|url| async move {
            client
                .get(url, None, None, Some(3), None)
                .await
                .unwrap()
                .status
                .as_u16()
        })
        .await;
    }

    #[rstest]
    fn raw_clients_share_rest_limit_for_one_route() {
        let mut first =
            HyperliquidRawHttpClient::new(HyperliquidEnvironment::Testnet, 60, None).unwrap();
        let mut second =
            HyperliquidRawHttpClient::new(HyperliquidEnvironment::Testnet, 60, None).unwrap();
        first.set_base_info_url("https://shared-http-limit.test/info".to_string());
        second.set_base_exchange_url("https://shared-http-limit.test/exchange".to_string());

        assert!(Arc::ptr_eq(&first.info_limiter, &second.exchange_limiter));
    }

    #[rstest]
    #[case::seconds("3", Some(3_000))]
    #[case::zero("0", Some(0))]
    #[case::past_date("Sun, 06 Nov 1994 08:49:36 GMT", Some(0))]
    #[case::overflow("18446744073709551615", Some(u64::MAX))]
    #[case::invalid("soon", None)]
    #[case::negative("-1", None)]
    fn retry_after_ms_parses_header(#[case] value: &str, #[case] expected: Option<u64>) {
        let headers = HashMap::from([(RETRY_AFTER_HEADER.to_string(), value.to_string())]);

        assert_eq!(HyperliquidRawHttpClient::retry_after_ms(&headers), expected);
    }

    #[rstest]
    #[case::transport(RetryFailure::Transport)]
    #[case::timeout(RetryFailure::Timeout)]
    #[case::request_timeout(RetryFailure::Response(408))]
    #[case::rate_limit(RetryFailure::Response(429))]
    #[case::server_error(RetryFailure::Response(500))]
    #[case::unavailable(RetryFailure::Response(503))]
    #[tokio::test]
    async fn info_retry_recovers(#[case] failure: RetryFailure) {
        let (addr, count, server) = start_retry_server(failure, 1, None).await;
        let mut client =
            HyperliquidRawHttpClient::new(HyperliquidEnvironment::Testnet, 1, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));

        let result = client
            .send_info_request_raw(&InfoRequest::meta())
            .await
            .unwrap();
        let expected: Value =
            serde_json::from_str(include_str!("../../test_data/http_meta_perp_sample.json"))
                .unwrap();

        server.abort();
        assert_eq!(result, expected);
        assert_eq!(count.load(Ordering::Relaxed), 2);
    }

    #[rstest]
    #[case::transport(RetryFailure::Transport)]
    #[case::timeout(RetryFailure::Timeout)]
    #[case::request_timeout(RetryFailure::Response(408))]
    #[case::rate_limit(RetryFailure::Response(429))]
    #[case::server_error(RetryFailure::Response(503))]
    #[tokio::test]
    async fn info_retry_exhausts_budget(#[case] failure: RetryFailure) {
        let (addr, count, server) = start_retry_server(failure, usize::MAX, Some("0")).await;
        let mut client =
            HyperliquidRawHttpClient::new(HyperliquidEnvironment::Testnet, 1, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));

        let error = client.info_meta().await.unwrap_err();

        server.abort();
        assert_retry_failure(error, failure, ("info", 20, Some(0)));
        assert_eq!(count.load(Ordering::Relaxed), 4);
    }

    #[rstest]
    #[case::bad_request(RetryFailure::Response(400))]
    #[case::unauthorized(RetryFailure::Response(401))]
    #[case::forbidden(RetryFailure::Response(403))]
    #[case::unprocessable(RetryFailure::Response(422))]
    #[case::invalid_json(RetryFailure::InvalidJson)]
    #[tokio::test]
    async fn info_retry_rejects_permanent_failures(#[case] failure: RetryFailure) {
        let (addr, count, server) = start_retry_server(failure, usize::MAX, Some("0")).await;
        let mut client =
            HyperliquidRawHttpClient::new(HyperliquidEnvironment::Testnet, 1, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));

        let error = client.info_meta().await.unwrap_err();

        server.abort();
        assert_retry_failure(error, failure, ("info", 20, Some(0)));
        assert_eq!(count.load(Ordering::Relaxed), 1);
    }

    #[rstest]
    #[case(0)]
    #[case(1)]
    #[case(2)]
    #[tokio::test]
    async fn info_retry_uses_configured_budget(#[case] max_retries: u32) {
        let failure = RetryFailure::Response(503);
        let (addr, count, server) = start_retry_server(failure, usize::MAX, Some("invalid")).await;
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Testnet, 1, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));
        client.set_info_retry_config(max_retries, Duration::ZERO, Duration::ZERO);

        let error = client.info_meta().await.unwrap_err();

        server.abort();
        assert_retry_failure(error, failure, ("info", 20, None));
        assert_eq!(count.load(Ordering::Relaxed), max_retries as usize + 1);
    }

    #[rstest]
    #[case::rate_limit(429)]
    #[case::unavailable(503)]
    #[tokio::test]
    async fn info_retry_honors_retry_after_above_backoff_cap(#[case] status: u16) {
        let (addr, count, server) =
            start_retry_server(RetryFailure::Response(status), 1, Some("1")).await;
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Testnet, 2, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));
        client.set_info_retry_config(1, Duration::ZERO, Duration::ZERO);
        let started = tokio::time::Instant::now();

        let result = client.info_meta().await;
        let elapsed = started.elapsed();

        server.abort();
        assert!(result.is_ok(), "{result:?}");
        assert_eq!(count.load(Ordering::Relaxed), 2);
        assert!(elapsed >= Duration::from_secs(1), "{elapsed:?}");
    }

    #[rstest]
    #[case::transport(RetryFailure::Transport)]
    #[case::timeout(RetryFailure::Timeout)]
    #[case::rate_limit(RetryFailure::Response(429))]
    #[case::server_error(RetryFailure::Response(503))]
    #[tokio::test]
    async fn exchange_does_not_retry_unknown_outcome(
        #[case] failure: RetryFailure,
        #[values("submit", "modify", "cancel")] operation: &str,
    ) {
        let (addr, count, server) = start_retry_server(failure, usize::MAX, Some("0")).await;
        let mut client = HyperliquidHttpClient::from_credentials(
            TEST_PRIVATE_KEY,
            None,
            HyperliquidEnvironment::Testnet,
            1,
            None,
        )
        .unwrap();
        client.set_base_exchange_url(format!("http://{addr}/exchange"));
        client.set_info_retry_config(9, Duration::ZERO, Duration::ZERO);
        let signing: Value =
            serde_json::from_str(include_str!("../../test_data/order_decimal_signing.json"))
                .unwrap();

        let action = match operation {
            "submit" => serde_json::from_value(signing["cases"][0]["action"].clone()).unwrap(),
            "modify" => HyperliquidExchangeAction::Modify {
                modify: serde_json::from_value(
                    signing["cases"][2]["action"]["modifies"][0].clone(),
                )
                .unwrap(),
            },
            "cancel" => {
                let cancels: Value =
                    serde_json::from_str(include_str!("../../test_data/cancel_actions.json"))
                        .unwrap();
                serde_json::from_value(cancels["cancel"]["normal"]["json"].clone()).unwrap()
            }
            _ => unreachable!(),
        };

        let error = client.post_action_exec(&action).await.unwrap_err();

        server.abort();

        assert_retry_failure(error, failure, ("exchange", 1, Some(0)));
        assert_eq!(count.load(Ordering::Relaxed), 1);
    }

    #[derive(Debug, Clone, Copy)]
    enum RetryFailure {
        Transport,
        Timeout,
        Response(u16),
        InvalidJson,
    }

    async fn start_retry_server(
        failure: RetryFailure,
        failures: usize,
        retry_after: Option<&'static str>,
    ) -> (SocketAddr, Arc<AtomicUsize>, tokio::task::JoinHandle<()>) {
        let count = Arc::new(AtomicUsize::new(0));
        let requests = count.clone();
        let handler = post(move |_body: axum::body::Bytes| {
            let count = requests.clone();
            async move {
                if count.fetch_add(1, Ordering::Relaxed) >= failures {
                    return include_str!("../../test_data/http_meta_perp_sample.json")
                        .into_response();
                }

                let mut response = match failure {
                    RetryFailure::Transport => {
                        let stream = futures_util::stream::once(async {
                            Err::<axum::body::Bytes, _>(std::io::Error::other("connection dropped"))
                        });

                        Response::new(axum::body::Body::from_stream(stream))
                    }
                    RetryFailure::Timeout => {
                        tokio::time::sleep(Duration::from_secs(2)).await;
                        "delayed response".into_response()
                    }
                    RetryFailure::Response(status) => {
                        (StatusCode::from_u16(status).unwrap(), "retry failure").into_response()
                    }
                    RetryFailure::InvalidJson => "invalid json".into_response(),
                };

                if let Some(value) = retry_after {
                    response
                        .headers_mut()
                        .insert(RETRY_AFTER_HEADER, value.parse().unwrap());
                }

                response
            }
        });

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let router = Router::new()
            .route("/info", handler.clone())
            .route("/exchange", handler);

        let server = tokio::spawn(async move {
            axum::serve(listener, router).await.unwrap();
        });

        (addr, count, server)
    }

    fn assert_retry_failure(
        error: Error,
        failure: RetryFailure,
        rate_limit: (&str, u32, Option<u64>),
    ) {
        match (failure, error) {
            (RetryFailure::Transport, Error::Transport(_))
            | (RetryFailure::Timeout, Error::Timeout)
            | (RetryFailure::InvalidJson, Error::Serde(_)) => {}
            (
                RetryFailure::Response(429),
                Error::RateLimit {
                    scope,
                    weight,
                    retry_after_ms,
                },
            ) => {
                assert_eq!((scope, weight, retry_after_ms), rate_limit);
            }
            (
                RetryFailure::Response(status),
                Error::Http {
                    status: actual,
                    message,
                },
            ) => {
                assert_eq!((actual, message.as_str()), (status, "retry failure"));
            }
            (_, e) => panic!("unexpected info failure: {e:?}"),
        }
    }

    fn perp_meta_with_assets(names: &[&str]) -> PerpMeta {
        PerpMeta {
            universe: names
                .iter()
                .map(|name| PerpAsset {
                    name: (*name).to_string(),
                    ..Default::default()
                })
                .collect(),
            margin_tables: Vec::new(),
            collateral_token: None,
        }
    }

    #[rstest]
    fn resolve_perp_dex_name_uses_empty_string_for_default_dex() {
        let meta = perp_meta_with_assets(&["BTC", "ETH"]);
        assert_eq!(resolve_perp_dex_name(0, &meta, None), "");
    }

    #[rstest]
    fn resolve_perp_dex_name_prefers_perp_dexs_entry() {
        let meta = perp_meta_with_assets(&["xyz:TSLA"]);
        let perp_dexs = vec![
            None,
            Some(PerpDex {
                name: "xyz".to_string(),
            }),
        ];
        assert_eq!(resolve_perp_dex_name(1, &meta, Some(&perp_dexs)), "xyz");
    }

    #[rstest]
    fn resolve_perp_dex_name_infers_from_asset_name_when_perp_dexs_missing() {
        let meta = perp_meta_with_assets(&["abc:TSLA", "abc:NVDA"]);
        assert_eq!(resolve_perp_dex_name(1, &meta, None), "abc");
    }

    #[rstest]
    fn test_build_submit_order_report_elides_waiting_for_fill_tag() {
        // A `Tag` status (for example the `waitingForFill` trigger child of a
        // `normalTpsl` bracket) must surface as `Ok(None)` so the caller leaves
        // the order SUBMITTED until the user-events stream confirms a real oid.
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_account_id(AccountId::from("HYPERLIQUID-001"));

        let response: HyperliquidExchangeResponse = serde_json::from_value(json!({
            "status": "ok",
            "response": {
                "type": "order",
                "data": {
                    "statuses": ["waitingForFill"]
                }
            }
        }))
        .unwrap();

        let result = client
            .build_submit_order_report(
                InstrumentId::from("ARB-USD-PERP.HYPERLIQUID"),
                ClientOrderId::from("O-WAITING-CHILD"),
                OrderSide::Buy,
                OrderType::StopMarket,
                Quantity::from("100"),
                TimeInForce::Gtc,
                None,
                Some(Price::from("0.16136")),
                response,
            )
            .unwrap();

        assert!(
            result.is_none(),
            "Tag status must elide so the order stays SUBMITTED, was {result:?}"
        );
    }

    #[rstest]
    fn test_build_submit_order_report_filled_uses_total_sz_decimal() {
        // An atomic `filled` submit response must surface as a FILLED report
        // carrying the venue oid and the total filled size built from the
        // Decimal `totalSz` at the instrument's size precision.
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_account_id(AccountId::from("HYPERLIQUID-001"));

        let base = Currency::new("ARB", 8, 0, "ARB", CurrencyType::Crypto);
        let usd = Currency::new("USD", 8, 0, "USD", CurrencyType::Crypto);
        let usdc = Currency::new("USDC", 6, 0, "USDC", CurrencyType::Crypto);
        let clock = get_atomic_clock_realtime();
        let ts = clock.get_time_ns();
        let perp = InstrumentAny::CryptoPerpetual(
            CryptoPerpetual::builder()
                .instrument_id(InstrumentId::new(
                    Symbol::new("ARB-USD-PERP"),
                    *HYPERLIQUID_VENUE,
                ))
                .raw_symbol(Symbol::new("ARB"))
                .base_currency(base)
                .quote_currency(usd)
                .settlement_currency(usdc)
                .is_inverse(false)
                .price_precision(5)
                .size_precision(2)
                .price_increment(Price::from("0.00001"))
                .size_increment(Quantity::from("0.01"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );
        client.cache_instrument(&perp);

        let response: HyperliquidExchangeResponse = serde_json::from_value(json!({
            "status": "ok",
            "response": {
                "type": "order",
                "data": {
                    "statuses": [{
                        "filled": {"totalSz": "0.5", "avgPx": "1.2345", "oid": 778899}
                    }]
                }
            }
        }))
        .unwrap();

        let report = client
            .build_submit_order_report(
                InstrumentId::from("ARB-USD-PERP.HYPERLIQUID"),
                ClientOrderId::from("O-FILLED-001"),
                OrderSide::Buy,
                OrderType::Market,
                Quantity::from("0.5"),
                TimeInForce::Ioc,
                None,
                None,
                response,
            )
            .unwrap()
            .expect("filled status must produce a report");

        assert_eq!(report.order_status, OrderStatus::Filled);
        assert_eq!(report.venue_order_id.as_str(), "778899");
        assert_eq!(report.filled_qty.as_decimal(), dec!(0.5));
    }

    #[derive(Clone, Default)]
    struct OutcomeMetaServerState {
        last_request_body: Arc<tokio::sync::Mutex<Option<Value>>>,
    }

    async fn handle_outcome_meta_info(
        State(state): State<OutcomeMetaServerState>,
        body: axum::body::Bytes,
    ) -> Response {
        let Ok(request_body): Result<Value, _> = serde_json::from_slice(&body) else {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "Invalid JSON body"})),
            )
                .into_response();
        };

        *state.last_request_body.lock().await = Some(request_body.clone());

        if request_body.get("type").and_then(|value| value.as_str()) != Some("outcomeMeta") {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "Expected outcomeMeta request"})),
            )
                .into_response();
        }

        Json(json!({
            "outcomes": [
                {
                    "outcome": 123,
                    "quoteToken": "USDC",
                    "name": "Recurring",
                    "description": "class:priceBinary|underlying:HYPE|expiry:20260310-1100|targetPrice:34.5|period:3m",
                    "sideSpecs": [
                        {"name": "Yes"},
                        {"name": "No"}
                    ]
                }
            ]
        }))
        .into_response()
    }

    async fn start_outcome_meta_server(state: OutcomeMetaServerState) -> SocketAddr {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let router = Router::new()
            .route("/info", post(handle_outcome_meta_info))
            .with_state(state);

        tokio::spawn(async move {
            axum::serve(listener, router).await.unwrap();
        });

        addr
    }

    async fn handle_unresolved_collateral_info(body: axum::body::Bytes) -> Response {
        let Ok(request_body): Result<Value, _> = serde_json::from_slice(&body) else {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "Invalid JSON body"})),
            )
                .into_response();
        };

        match request_body.get("type").and_then(|value| value.as_str()) {
            Some("spotMeta") => (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "spot metadata unavailable"})),
            )
                .into_response(),
            Some("allPerpMetas") => Json(json!([
                {
                    "collateralToken": 360,
                    "marginTables": [],
                    "universe": [
                        {
                            "maxLeverage": 20,
                            "name": "km:US500",
                            "szDecimals": 3
                        }
                    ]
                }
            ]))
            .into_response(),
            _ => Json(json!({"universe": [], "marginTables": []})).into_response(),
        }
    }

    async fn start_unresolved_collateral_server() -> SocketAddr {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let router = Router::new().route("/info", post(handle_unresolved_collateral_info));

        tokio::spawn(async move {
            axum::serve(listener, router).await.unwrap();
        });

        addr
    }

    #[rstest]
    fn stable_json_roundtrips() {
        let v = serde_json::json!({"type":"l2Book","coin":"BTC"});
        let s = serde_json::to_string(&v).unwrap();
        // Parse back to ensure JSON structure is correct, regardless of field order
        let parsed: serde_json::Value = serde_json::from_str(&s).unwrap();
        assert_eq!(parsed["type"], "l2Book");
        assert_eq!(parsed["coin"], "BTC");
        assert_eq!(parsed, v);
    }

    #[rstest]
    fn info_pretty_shape() {
        let r = InfoRequest::l2_book("BTC");
        let val = serde_json::to_value(&r).unwrap();
        let pretty = serde_json::to_string_pretty(&val).unwrap();
        assert!(pretty.contains("\"type\": \"l2Book\""));
        assert!(pretty.contains("\"coin\": \"BTC\""));
    }

    #[rstest]
    fn test_client_order_id_cloid_cache_is_stable_and_first_write_wins() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        let client_order_id = ClientOrderId::new("O-CLOID-CACHE");
        let other_client_order_id = ClientOrderId::new("O-CLOID-CACHE-OTHER");
        let duplicate_client_order_id = ClientOrderId::new("O-CLOID-CACHE-DUPLICATE");
        let explicit_cloid = Cloid::from_hex("0x1234567890abcdef1234567890abcdef").unwrap();

        let first = client.get_or_generate_client_order_id_cloid(client_order_id);
        let second = client.get_or_generate_client_order_id_cloid(client_order_id);
        client.cache_client_order_id_cloid(client_order_id, explicit_cloid);
        client.cache_client_order_id_cloid(other_client_order_id, explicit_cloid);
        client.cache_client_order_id_cloid(duplicate_client_order_id, explicit_cloid);

        assert_eq!(first, Cloid::from_client_order_id(client_order_id));
        assert_eq!(first, second);
        assert_eq!(
            client.cached_client_order_id_cloid(&client_order_id),
            Some(first),
            "cache insert must not overwrite an existing generated CLOID",
        );
        assert_eq!(
            client.unique_cached_client_order_id_cloid(&client_order_id),
            Some(first),
        );
        assert_eq!(
            client.cached_client_order_id_cloid(&other_client_order_id),
            Some(explicit_cloid),
        );
        assert_eq!(
            client.unique_cached_client_order_id_cloid(&other_client_order_id),
            None,
            "duplicate CLOID mappings are not safe modify targets",
        );
        assert_eq!(
            client.remove_client_order_id_cloid(&client_order_id),
            Some(first),
        );
        assert_eq!(client.cached_client_order_id_cloid(&client_order_id), None);
    }

    #[rstest]
    fn test_builder_attribution_defaults_to_mainnet_builder() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        let builder = client
            .builder_attribution()
            .expect("mainnet client should include builder attribution by default");

        assert!(client.include_builder_attribution());
        assert_eq!(builder.address, NAUTILUS_BUILDER_ADDRESS);
        assert_eq!(builder.fee_tenths_bp, 0);
    }

    #[rstest]
    fn test_builder_attribution_disabled_returns_none() {
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_include_builder_attribution(false);

        assert!(!client.include_builder_attribution());
        assert!(client.builder_attribution().is_none());
    }

    #[rstest]
    fn test_builder_attribution_omitted_on_testnet() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Testnet, 60, None).unwrap();

        assert!(client.include_builder_attribution());
        assert!(client.builder_attribution().is_none());
    }

    #[rstest]
    #[tokio::test]
    async fn test_production_client_get_outcome_meta_uses_outcome_meta_request() {
        let state = OutcomeMetaServerState::default();
        let addr = start_outcome_meta_server(state.clone()).await;
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));

        let meta = client.get_outcome_meta().await.unwrap();
        let request_body = state.last_request_body.lock().await.clone().unwrap();

        assert_eq!(request_body, json!({"type": "outcomeMeta"}));
        assert_eq!(meta.outcomes.len(), 1);
        assert_eq!(meta.outcomes[0].outcome, 123);
        assert_eq!(meta.outcomes[0].quote_token.as_deref(), Some("USDC"));
        assert_eq!(meta.outcomes[0].name, "Recurring");
        assert_eq!(meta.outcomes[0].side_specs.len(), 2);
        assert_eq!(meta.outcomes[0].side_specs[0].name, "Yes");
        assert_eq!(meta.outcomes[0].side_specs[1].name, "No");
    }

    #[rstest]
    #[tokio::test]
    async fn test_request_instrument_defs_errors_when_non_usdc_collateral_unresolved() {
        let addr = start_unresolved_collateral_server().await;
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_base_info_url(format!("http://{addr}/info"));

        let err = client.request_instrument_defs().await.unwrap_err();

        assert_eq!(
            err.to_string(),
            "decode error: failed to resolve perp settlement currency for dex 0: \
             Spot metadata required to resolve perp collateral token 360",
        );
    }

    #[rstest]
    fn test_with_credentials_preserves_explicit_account_address() {
        let account_address = "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        let client = HyperliquidHttpClient::with_credentials(
            Some(TEST_PRIVATE_KEY.to_string()),
            None,
            Some(account_address),
            HyperliquidEnvironment::Mainnet,
            60,
            None,
        )
        .unwrap();

        assert_eq!(client.get_account_address().unwrap(), account_address);
    }

    #[rstest]
    fn test_from_resolved_credentials_preserves_account_address_without_private_key() {
        let account_address = "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        let client = HyperliquidHttpClient::from_resolved_credentials(
            None,
            None,
            Some(account_address.to_string()),
            HyperliquidEnvironment::Mainnet,
            60,
            None,
        )
        .unwrap();

        assert_eq!(client.get_account_address().unwrap(), account_address);
    }

    #[rstest]
    fn test_with_credentials_rejects_invalid_account_address() {
        let e = HyperliquidHttpClient::with_credentials(
            Some(TEST_PRIVATE_KEY.to_string()),
            None,
            Some("0xinvalid"),
            HyperliquidEnvironment::Mainnet,
            60,
            Some("invalid proxy URL".to_string()),
        )
        .unwrap_err();

        assert_eq!(
            e.to_string(),
            "bad request: Account address must be 20 bytes of valid hex"
        );
    }

    #[rstest]
    fn test_set_account_address_normalizes_and_preserves_override_on_error() {
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client
            .set_account_address(Some(
                " \t0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045 ".to_string(),
            ))
            .unwrap();
        let e = client
            .set_account_address(Some("0xinvalid".to_string()))
            .unwrap_err();

        assert_eq!(
            e.to_string(),
            "bad request: Account address must be 20 bytes of valid hex"
        );
        assert_eq!(
            client.get_account_address().unwrap(),
            "0xd8da6bf26964af9d7eed9e03e53415d37aa96045"
        );
    }

    #[rstest]
    #[case(None)]
    #[case(Some("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"))]
    fn test_set_account_address_clears_override(#[case] vault_address: Option<&str>) {
        let mut client = HyperliquidHttpClient::from_resolved_credentials(
            Some(TEST_PRIVATE_KEY.to_string()),
            vault_address,
            Some("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string()),
            HyperliquidEnvironment::Mainnet,
            60,
            None,
        )
        .unwrap();

        let expected_address = match vault_address {
            Some(vault) => vault.to_string(),
            None => client.get_user_address().unwrap(),
        };

        client.set_account_address(None).unwrap();

        assert_eq!(client.account_address, None);
        assert_eq!(client.get_account_address().unwrap(), expected_address);
    }

    #[rstest]
    fn test_cache_instrument_by_raw_symbol() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        // Create a test instrument with base currency "vntls:vCURSOR"
        let base_code = "vntls:vCURSOR";
        let quote_code = "USDC";

        // Register the custom currency
        {
            let mut currency_map = CURRENCY_MAP.lock();
            if !currency_map.contains_key(base_code) {
                currency_map.insert(
                    base_code.to_string(),
                    Currency::new(base_code, 8, 0, base_code, CurrencyType::Crypto),
                );
            }
        }

        let base_currency = Currency::new(base_code, 8, 0, base_code, CurrencyType::Crypto);
        let quote_currency = Currency::new(quote_code, 6, 0, quote_code, CurrencyType::Crypto);

        // Nautilus symbol is "vntls:vCURSOR-USDC-SPOT"
        let symbol = Symbol::new("vntls:vCURSOR-USDC-SPOT");
        let venue = *HYPERLIQUID_VENUE;
        let instrument_id = InstrumentId::new(symbol, venue);

        // raw_symbol is set to the base currency "vntls:vCURSOR" (see parse.rs)
        let raw_symbol = Symbol::new(base_code);

        let clock = get_atomic_clock_realtime();
        let ts = clock.get_time_ns();

        let instrument = InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(instrument_id)
                .raw_symbol(raw_symbol)
                .base_currency(base_currency)
                .quote_currency(quote_currency)
                .price_precision(8)
                .size_precision(8)
                .price_increment(Price::from("0.00000001"))
                .size_increment(Quantity::from("0.00000001"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );

        // Cache the instrument
        client.cache_instrument(&instrument);

        // Verify it can be looked up by full symbol
        let instruments = client.instruments.load();
        let by_full_symbol = instruments.get(&Ustr::from("vntls:vCURSOR-USDC-SPOT"));
        assert!(
            by_full_symbol.is_some(),
            "Instrument should be accessible by full symbol"
        );
        assert_eq!(by_full_symbol.unwrap().id(), instrument.id());

        // Verify it can be looked up by raw_symbol (coin) - backward compatibility
        let by_raw_symbol = instruments.get(&Ustr::from("vntls:vCURSOR"));
        assert!(
            by_raw_symbol.is_some(),
            "Instrument should be accessible by raw_symbol (Hyperliquid coin identifier)"
        );
        assert_eq!(by_raw_symbol.unwrap().id(), instrument.id());
        drop(instruments);

        // Verify it can be looked up by composite key (coin, product_type)
        let instruments_by_coin = client.instruments_by_coin.load();
        let by_coin =
            instruments_by_coin.get(&(Ustr::from("vntls:vCURSOR"), HyperliquidProductType::Spot));
        assert!(
            by_coin.is_some(),
            "Instrument should be accessible by coin and product type"
        );
        assert_eq!(by_coin.unwrap().id(), instrument.id());
        drop(instruments_by_coin);

        // Verify get_or_create_instrument works with product type
        let retrieved_with_type = client.get_or_create_instrument(
            &Ustr::from("vntls:vCURSOR"),
            Some(HyperliquidProductType::Spot),
        );
        assert!(retrieved_with_type.is_some());
        assert_eq!(retrieved_with_type.unwrap().id(), instrument.id());

        // Verify get_or_create_instrument works without product type (fallback)
        let retrieved_without_type =
            client.get_or_create_instrument(&Ustr::from("vntls:vCURSOR"), None);
        assert!(retrieved_without_type.is_some());
        assert_eq!(retrieved_without_type.unwrap().id(), instrument.id());
    }

    #[rstest]
    fn test_get_or_create_instrument_outcome_fallback_no_product_type() {
        // HTTP fill payloads for HIP-4 outcomes arrive with `coin = "#E"` and
        // no product-type context, so the no-product fallback in
        // `get_or_create_instrument` must check the Outcome bucket. Without
        // this, venue Settlement and userOutcome fills are silently dropped
        // from request_fill_reports / request_order_status_reports.
        use nautilus_core::time::get_atomic_clock_realtime;
        use nautilus_model::{
            enums::AssetClass,
            identifiers::{InstrumentId, Symbol},
            instruments::{BinaryOption, InstrumentAny},
            types::{Currency, Price, Quantity},
        };

        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        let coin = "#500";
        let token = "+500";

        let usdh = Currency::new("USDH", 8, 0, "Hyperliquid USD", CurrencyType::Crypto);
        let symbol = Symbol::new(token);
        let raw_symbol = Symbol::new(coin);
        let venue = *HYPERLIQUID_VENUE;
        let instrument_id = InstrumentId::new(symbol, venue);

        let clock = get_atomic_clock_realtime();
        let ts = clock.get_time_ns();

        let binary = InstrumentAny::BinaryOption(
            BinaryOption::builder()
                .instrument_id(instrument_id)
                .raw_symbol(raw_symbol)
                .asset_class(AssetClass::Alternative)
                .currency(usdh)
                .activation_ns(Default::default())
                .expiration_ns(Default::default())
                .price_precision(4)
                .size_precision(2)
                .price_increment(Price::from("0.0001"))
                .size_increment(Quantity::from("0.01"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );

        client.cache_instrument(&binary);

        let with_type = client
            .get_or_create_instrument(&Ustr::from(coin), Some(HyperliquidProductType::Outcome));
        assert!(with_type.is_some());
        let with_type = with_type.unwrap();
        assert_eq!(with_type.id(), instrument_id);
        assert_eq!(with_type.quote_currency(), usdh);

        let no_type = client.get_or_create_instrument(&Ustr::from(coin), None);
        assert!(
            no_type.is_some(),
            "Outcome coin must resolve through the no-product fallback",
        );
        let no_type = no_type.unwrap();
        assert_eq!(no_type.id(), instrument_id);
        assert_eq!(no_type.quote_currency(), usdh);

        let missing = client.get_or_create_instrument(&Ustr::from("#9999"), None);
        assert!(missing.is_none());
    }

    #[rstest]
    #[case::coin_form("#200", None, "20-YES-OUTCOME", "#200", 100_000_200)]
    #[case::token_form(
        "+201",
        Some(HyperliquidProductType::Outcome),
        "20-NO-OUTCOME",
        "#201",
        100_000_201
    )]
    fn test_get_or_create_instrument_resolves_unlisted_outcome(
        #[case] coin: &str,
        #[case] product_type: Option<HyperliquidProductType>,
        #[case] symbol: &str,
        #[case] raw_symbol: &str,
        #[case] asset_index: u32,
    ) {
        // Settled outcomes drop out of `outcomeMeta`, so fills carry `#E` coins
        // and spot balances carry `+E` tokens for markets never loaded
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        let instrument = client
            .get_or_create_instrument(&Ustr::from(coin), product_type)
            .expect("unlisted outcome coin must resolve from its encoding");

        assert_eq!(
            instrument.id(),
            InstrumentId::new(Symbol::new(symbol), *HYPERLIQUID_VENUE)
        );
        assert_eq!(instrument.raw_symbol(), Symbol::new(raw_symbol));
        assert_eq!(instrument.price_precision(), 5);
        assert_eq!(instrument.size_precision(), 0);
        assert_eq!(instrument.quote_currency(), Currency::USDC());
        assert_eq!(instrument.settlement_currency(), Currency::USDC());
        assert_eq!(client.get_asset_index(symbol), Some(asset_index));
    }

    #[rstest]
    fn test_cache_instrument_base_alias_first_write_wins_for_spot() {
        // Two spot pairs share the base token "HYPE": the canonical pair is
        // cached first; a subsequent non-canonical pair must not overwrite the
        // base-token alias so lookups by "HYPE" keep resolving to the canonical
        // instrument.
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        let hype = Currency::new("HYPE", 8, 0, "HYPE", CurrencyType::Crypto);
        let usdc = Currency::new("USDC", 6, 0, "USDC", CurrencyType::Crypto);
        let clock = get_atomic_clock_realtime();
        let ts = clock.get_time_ns();

        let canonical = InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(InstrumentId::new(
                    Symbol::new("HYPE-USDC-SPOT"),
                    *HYPERLIQUID_VENUE,
                ))
                .raw_symbol(Symbol::new("@107"))
                .base_currency(hype)
                .quote_currency(usdc)
                .price_precision(5)
                .size_precision(2)
                .price_increment(Price::from("0.00001"))
                .size_increment(Quantity::from("0.01"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );

        let non_canonical = InstrumentAny::CurrencyPair(
            CurrencyPair::builder()
                .instrument_id(InstrumentId::new(
                    Symbol::new("HYPE-USDC-SPOT"),
                    *HYPERLIQUID_VENUE,
                ))
                .raw_symbol(Symbol::new("@999"))
                .base_currency(hype)
                .quote_currency(usdc)
                .price_precision(5)
                .size_precision(2)
                .price_increment(Price::from("0.00001"))
                .size_increment(Quantity::from("0.01"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );

        client.merge_instruments(&[canonical.clone(), non_canonical.clone()]);

        let instruments_by_coin = client.instruments_by_coin.load();
        let by_base = instruments_by_coin
            .get(&(Ustr::from("HYPE"), HyperliquidProductType::Spot))
            .expect("base alias must resolve");
        assert_eq!(
            by_base.raw_symbol().inner().as_str(),
            "@107",
            "base alias must point to the canonical pair, not the one cached later",
        );
        drop(instruments_by_coin);

        let InstrumentAny::CurrencyPair(mut refreshed) = canonical else {
            unreachable!();
        };

        refreshed.price_precision = 4;
        refreshed.price_increment = Price::from("0.0001");
        refreshed.size_precision = 3;
        refreshed.size_increment = Quantity::from("0.001");
        let refreshed = InstrumentAny::CurrencyPair(refreshed);
        client.merge_instruments(&[refreshed.clone(), non_canonical]);

        let resolved = client
            .get_or_create_instrument(&Ustr::from("HYPE"), Some(HyperliquidProductType::Spot))
            .unwrap();
        assert_eq!(
            serde_json::to_value(&resolved).unwrap(),
            serde_json::to_value(&refreshed).unwrap(),
        );
    }

    #[rstest]
    fn test_cache_instrument_perp_aliases_sanitized_base() {
        // HIP-3 perp with wildcard-bearing venue name: `instrument_id.symbol`
        // is sanitized but order paths derive a coin key by splitting that
        // sanitized symbol on `-`. The cache must alias on the sanitized base
        // so those lookups resolve to the same instrument cached under
        // `raw_symbol` (the venue-official name).
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        let base_currency = Currency::new(
            "dex:STREAMABCD****",
            8,
            0,
            "dex:STREAMABCD****",
            CurrencyType::Crypto,
        );
        let usd = Currency::new("USD", 8, 0, "USD", CurrencyType::Crypto);
        let usdc = Currency::new("USDC", 6, 0, "USDC", CurrencyType::Crypto);
        let clock = get_atomic_clock_realtime();
        let ts = clock.get_time_ns();

        let hip3 = InstrumentAny::CryptoPerpetual(
            CryptoPerpetual::builder()
                .instrument_id(InstrumentId::new(
                    Symbol::new("dex:STREAMABCDxxxx-USD-PERP"),
                    *HYPERLIQUID_VENUE,
                ))
                .raw_symbol(Symbol::new("dex:STREAMABCD****"))
                .base_currency(base_currency)
                .quote_currency(usd)
                .settlement_currency(usdc)
                .is_inverse(false)
                .price_precision(6)
                .size_precision(3)
                .price_increment(Price::from("0.000001"))
                .size_increment(Quantity::from("0.001"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );

        client.cache_instrument(&hip3);

        let instruments_by_coin = client.instruments_by_coin.load();
        let by_raw = instruments_by_coin
            .get(&(
                Ustr::from("dex:STREAMABCD****"),
                HyperliquidProductType::Perp,
            ))
            .expect("venue coin lookup must resolve");
        assert_eq!(by_raw.id(), hip3.id());

        let by_sanitized = instruments_by_coin
            .get(&(
                Ustr::from("dex:STREAMABCDxxxx"),
                HyperliquidProductType::Perp,
            ))
            .expect("sanitized base lookup must resolve");
        assert_eq!(by_sanitized.id(), hip3.id());
        drop(instruments_by_coin);

        // Confirm the order-submission lookup path resolves through the alias.
        let resolved = client
            .get_or_create_instrument(
                &Ustr::from("dex:STREAMABCDxxxx"),
                Some(HyperliquidProductType::Perp),
            )
            .expect("get_or_create_instrument must resolve sanitized base for HIP-3");
        assert_eq!(resolved.id(), hip3.id());

        let InstrumentAny::CryptoPerpetual(mut refreshed) = hip3 else {
            unreachable!();
        };

        refreshed.price_precision = 4;
        refreshed.price_increment = Price::from("0.0001");
        refreshed.size_precision = 2;
        refreshed.size_increment = Quantity::from("0.01");
        let refreshed = InstrumentAny::CryptoPerpetual(refreshed);
        client.cache_instrument(&refreshed);

        let resolved = client
            .get_or_create_instrument(
                &Ustr::from("dex:STREAMABCDxxxx"),
                Some(HyperliquidProductType::Perp),
            )
            .unwrap();
        assert_eq!(
            serde_json::to_value(&resolved).unwrap(),
            serde_json::to_value(&refreshed).unwrap(),
        );
    }

    #[tokio::test]
    #[ignore = "connects to Hyperliquid mainnet public instrument metadata"]
    async fn test_live_mainnet_instrument_refresh_updates_aliases() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        let original = client.request_instruments().await.unwrap();
        assert!(!original.is_empty());

        for instrument in original {
            client.cache_instrument(&instrument);
        }

        let refreshed = client.request_instruments().await.unwrap();
        let mut spot_aliases = 0;
        let mut perp_aliases = 0;

        for instrument in &refreshed {
            client.cache_instrument(instrument);
            let symbol = instrument.symbol();
            let product = HyperliquidProductType::from_symbol(symbol.as_str()).unwrap();
            let alias = super::cache_alias_for_symbol(symbol.as_str()).unwrap();
            let resolved = client
                .get_or_create_instrument(&Ustr::from(alias.as_str()), Some(product))
                .unwrap();

            if resolved.raw_symbol() != instrument.raw_symbol() {
                continue;
            }

            assert_eq!(
                serde_json::to_value(&resolved).unwrap(),
                serde_json::to_value(instrument).unwrap(),
            );
            assert_eq!(
                client.get_price_precision(symbol.as_str()),
                Some(instrument.price_precision()),
            );

            if alias != instrument.raw_symbol().as_str() {
                match product {
                    HyperliquidProductType::Spot => spot_aliases += 1,
                    HyperliquidProductType::Perp => perp_aliases += 1,
                    HyperliquidProductType::Outcome => {}
                }
            }
        }

        assert!(spot_aliases > 0);
        println!(
            "Validated {} refreshed definitions, {spot_aliases} spot aliases, and {perp_aliases} perp aliases",
            refreshed.len()
        );
    }

    fn perp_with_asset_index(symbol: &str, asset_index: Option<u32>) -> InstrumentAny {
        let base_currency = Currency::new("NEW", 8, 0, "NEW", CurrencyType::Crypto);
        let usd = Currency::new("USD", 8, 0, "USD", CurrencyType::Crypto);
        let usdc = Currency::new("USDC", 6, 0, "USDC", CurrencyType::Crypto);
        let ts = get_atomic_clock_realtime().get_time_ns();

        let info = asset_index.map(|asset_index| {
            Params::new([(ASSET_INDEX_INFO_KEY.to_string(), asset_index.into())].into())
        });

        InstrumentAny::CryptoPerpetual(
            CryptoPerpetual::builder()
                .instrument_id(InstrumentId::new(Symbol::new(symbol), *HYPERLIQUID_VENUE))
                .raw_symbol(Symbol::new("NEW"))
                .base_currency(base_currency)
                .quote_currency(usd)
                .settlement_currency(usdc)
                .is_inverse(false)
                .price_precision(6)
                .size_precision(3)
                .price_increment(Price::from("0.000001"))
                .size_increment(Quantity::from("0.001"))
                .maybe_info(info)
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        )
    }

    #[rstest]
    fn test_cache_instrument_registers_asset_index_from_info() {
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        client.cache_instrument(&perp_with_asset_index("NEW-USD-PERP", Some(42)));

        assert_eq!(client.get_asset_index("NEW-USD-PERP"), Some(42));
    }

    #[rstest]
    fn test_cache_instrument_without_asset_index_info_retains_existing_index() {
        // Guessing an index would route orders to the wrong asset, so an
        // instrument missing the info key must leave the map untouched.
        let client = HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();

        client.cache_instrument(&perp_with_asset_index("NEW-USD-PERP", Some(42)));
        let asset_indices = client.asset_indices.load();
        client.cache_instrument(&perp_with_asset_index("NEW-USD-PERP", None));
        client.cache_instrument(&perp_with_asset_index("OTHER-USD-PERP", None));

        assert_eq!(client.get_asset_index("NEW-USD-PERP"), Some(42));
        assert_eq!(client.get_asset_index("OTHER-USD-PERP"), None);
        assert!(Arc::ptr_eq(&asset_indices, &client.asset_indices.load()));
    }

    fn btc_history_client() -> HyperliquidHttpClient {
        let mut client =
            HyperliquidHttpClient::new(HyperliquidEnvironment::Mainnet, 60, None).unwrap();
        client.set_account_id(AccountId::from("HYPERLIQUID-001"));

        let base = Currency::new("BTC", 8, 0, "BTC", CurrencyType::Crypto);
        let usd = Currency::new("USD", 8, 0, "USD", CurrencyType::Crypto);
        let usdc = Currency::new("USDC", 6, 0, "USDC", CurrencyType::Crypto);
        let ts = get_atomic_clock_realtime().get_time_ns();
        let perp = InstrumentAny::CryptoPerpetual(
            CryptoPerpetual::builder()
                .instrument_id(InstrumentId::new(
                    Symbol::new("BTC-USD-PERP"),
                    *HYPERLIQUID_VENUE,
                ))
                .raw_symbol(Symbol::new("BTC"))
                .base_currency(base)
                .quote_currency(usd)
                .settlement_currency(usdc)
                .is_inverse(false)
                .price_precision(1)
                .size_precision(5)
                .price_increment(Price::from("0.1"))
                .size_increment(Quantity::from("0.00001"))
                .ts_event(ts)
                .ts_init(ts)
                .build()
                .unwrap(),
        );
        client.cache_instrument(&perp);
        client
    }

    fn history_rows(rows: Value) -> Vec<HyperliquidOrderStatusEntry> {
        serde_json::from_value(rows).unwrap()
    }

    // Builds an order from the report the way reconciliation does for an external order
    fn assert_initializes_order(report: &OrderStatusReport) -> OrderAny {
        let initialized = OrderInitialized::new_checked(
            TraderId::from("TESTER-001"),
            StrategyId::from("EXTERNAL"),
            report.instrument_id,
            ClientOrderId::from(report.venue_order_id.as_str()),
            report.order_side.expect("report has an order side"),
            report.order_type,
            report.quantity,
            report.time_in_force,
            report.post_only,
            report.reduce_only,
            false,
            true,
            UUID4::new(),
            UnixNanos::default(),
            UnixNanos::default(),
            report.price,
            report.activation_price,
            report.trigger_price,
            report.trigger_type,
            report.limit_offset,
            report.trailing_offset,
            report.trailing_offset_type,
            report.expire_time,
            report.display_qty,
            None,
            None,
            report.contingency_type,
            report.order_list_id,
            report.linked_order_ids.clone(),
            report.parent_order_id,
            None,
            None,
            None,
            None,
        )
        .expect("report must initialize an order");
        OrderAny::from_events(vec![OrderEventAny::Initialized(initialized)])
            .expect("order must replay from its initialization")
    }

    // `historicalOrders` rows for a position stop that triggered and filled: once filled, the
    // venue zeroes `triggerPx` and clears `isTrigger`, so only the earlier rows carry the trigger
    fn triggered_stop_rows() -> Value {
        let trigger_row = |status: &str, ts: u64| {
            json!({
                "order": {
                    "coin": "BTC", "side": "A", "limitPx": "58800.0", "sz": "0.0",
                    "oid": 900000001_u64, "timestamp": 1700000000000_u64,
                    "triggerCondition": "Price below 59500", "isTrigger": true,
                    "triggerPx": "59500.0", "children": [], "isPositionTpsl": true,
                    "reduceOnly": true, "orderType": "Stop Market", "origSz": "0.0", "tif": null,
                    "cloid": "0x00000000000000000000000000000a01"
                },
                "status": status,
                "statusTimestamp": ts
            })
        };
        json!([
            filled_stop_row(),
            trigger_row("triggered", 1700000600000),
            trigger_row("open", 1700000000000),
        ])
    }

    fn filled_stop_row() -> Value {
        json!({
            "order": {
                "coin": "BTC", "side": "A", "limitPx": "58200.0", "sz": "0.0",
                "oid": 900000001_u64, "timestamp": 1700000600000_u64,
                "triggerCondition": "Triggered", "isTrigger": false, "triggerPx": "0.0",
                "children": [], "isPositionTpsl": true, "reduceOnly": true,
                "orderType": "Stop Market", "origSz": "0.02", "tif": "Gtc",
                "cloid": "0x00000000000000000000000000000a01"
            },
            "status": "filled",
            "statusTimestamp": 1700000600000_u64
        })
    }

    #[rstest]
    fn test_historical_reports_triggered_stop_keeps_trigger_from_earlier_rows() {
        let client = btc_history_client();

        let sweep = client
            .historical_order_status_reports_from_response(
                history_rows(triggered_stop_rows()),
                None,
            )
            .unwrap();

        assert!(sweep.complete);
        assert_eq!(sweep.reports.len(), 1);
        let report = &sweep.reports[0];
        assert_eq!(report.venue_order_id, VenueOrderId::new("900000001"));
        assert_eq!(report.order_type, OrderType::StopMarket);
        assert_eq!(report.order_status, OrderStatus::Filled);
        assert_eq!(report.filled_qty, Quantity::from("0.02"));
        assert_eq!(report.trigger_price, Some(Price::from("59500.0")));
        assert_eq!(report.trigger_type, Some(TriggerType::Default));
        assert_initializes_order(report);
    }

    #[rstest]
    fn test_historical_reports_triggered_stop_limit_keeps_trigger_and_price() {
        let client = btc_history_client();
        let mut rows = triggered_stop_rows();
        for row in rows.as_array_mut().unwrap() {
            row["order"]["orderType"] = json!("Stop Limit");
        }

        let sweep = client
            .historical_order_status_reports_from_response(history_rows(rows), None)
            .unwrap();

        assert_eq!(sweep.reports.len(), 1);
        let report = &sweep.reports[0];
        assert_eq!(report.order_type, OrderType::StopLimit);
        // The filled row carries no price; the merge takes it from the earlier rows
        assert_eq!(report.price, Some(Price::from("58800.0")));
        assert_eq!(report.trigger_price, Some(Price::from("59500.0")));
        assert_eq!(report.trigger_type, Some(TriggerType::Default));
        assert_initializes_order(report);
    }

    #[rstest]
    #[case("Stop Market", "filled", OrderType::Market)]
    #[case("Take Profit Market", "filled", OrderType::Market)]
    #[case("Stop Limit", "filled", OrderType::Market)]
    #[case("Take Profit Limit", "filled", OrderType::Market)]
    #[case("Stop Market", "canceled", OrderType::Market)]
    #[case("Take Profit Market", "canceled", OrderType::Market)]
    #[case("Stop Limit", "canceled", OrderType::Limit)]
    #[case("Take Profit Limit", "canceled", OrderType::Limit)]
    fn test_historical_reports_closed_trigger_without_trigger_rows_is_untriggered(
        #[case] label: &str,
        #[case] status: &str,
        #[case] expected: OrderType,
    ) {
        // Only the zeroed final row survives (older rows aged out of the history window). A
        // filled row carries no price, so only a canceled limit-style row can stay a limit order.
        let client = btc_history_client();
        let mut row = filled_stop_row();
        row["order"]["orderType"] = json!(label);
        row["status"] = json!(status);
        // A canceled order's `sz` holds its unfilled size
        let expected_filled = if status == "canceled" {
            row["order"]["sz"] = json!("0.02");
            "0"
        } else {
            "0.02"
        };

        let sweep = client
            .historical_order_status_reports_from_response(history_rows(json!([row])), None)
            .unwrap();

        assert_eq!(sweep.reports.len(), 1);
        let report = &sweep.reports[0];
        assert_eq!(report.order_type, expected);
        assert_eq!(report.filled_qty, Quantity::from(expected_filled));
        assert!(report.trigger_price.is_none());
        assert!(report.trigger_type.is_none());
        assert_eq!(report.price.is_some(), expected == OrderType::Limit);
        assert!(report.reduce_only);
        assert_initializes_order(report);
    }

    #[rstest]
    fn test_historical_reports_triggered_stop_filled_with_remainder() {
        // Both fixes on one row: a triggered stop's zeroed final row that also left a remainder
        let client = btc_history_client();
        let mut row = filled_stop_row();
        row["order"]["sz"] = json!("0.0008");

        let sweep = client
            .historical_order_status_reports_from_response(history_rows(json!([row])), None)
            .unwrap();

        assert_eq!(sweep.reports.len(), 1);
        let report = &sweep.reports[0];
        assert_eq!(report.order_status, OrderStatus::Canceled);
        assert_eq!(report.order_type, OrderType::Market);
        assert_eq!(report.filled_qty, Quantity::from("0.0192"));
        assert!(report.price.is_none());
        assert_initializes_order(report);
    }

    #[rstest]
    fn test_historical_reports_open_trigger_without_trigger_price_keeps_order_type() {
        // An order that is still working keeps its trigger type even when its row lacks a
        // trigger price
        let client = btc_history_client();
        let mut row = filled_stop_row();
        row["status"] = json!("triggered");

        let sweep = client
            .historical_order_status_reports_from_response(history_rows(json!([row])), None)
            .unwrap();

        assert_eq!(sweep.reports.len(), 1);
        assert_eq!(sweep.reports[0].order_type, OrderType::StopMarket);
    }

    // An IOC order the venue marks `filled` although 0.0005 of 0.015 went unfilled: `sz`
    // carries the canceled remainder
    #[rstest]
    fn test_historical_reports_partially_filled_ioc_is_canceled() {
        let client = btc_history_client();
        let row = |status: &str, sz: &str| {
            json!({
                "order": {
                    "coin": "BTC", "side": "A", "limitPx": "60100.0", "sz": sz,
                    "oid": 900000002_u64, "timestamp": 1700001000000_u64,
                    "triggerCondition": "N/A", "isTrigger": false, "triggerPx": "0.0",
                    "children": [], "isPositionTpsl": false, "reduceOnly": false,
                    "orderType": "Limit", "origSz": "0.015", "tif": "Ioc",
                    "cloid": "0x00000000000000000000000000000a02"
                },
                "status": status,
                "statusTimestamp": 1700001000000_u64
            })
        };

        let sweep = client
            .historical_order_status_reports_from_response(
                history_rows(json!([row("filled", "0.0005"), row("open", "0.015")])),
                None,
            )
            .unwrap();

        assert_eq!(sweep.reports.len(), 1);
        let report = &sweep.reports[0];
        assert_eq!(report.order_status, OrderStatus::Canceled);
        assert_eq!(report.order_type, OrderType::Limit);
        assert_eq!(report.quantity, Quantity::from("0.015"));
        assert_eq!(report.filled_qty, Quantity::from("0.0145"));
        // The filled row carries no price; the merge takes the limit price from the open row
        assert_eq!(report.price, Some(Price::from("60100.0")));
        assert_eq!(
            report.cancel_reason.as_deref(),
            Some("Unfilled remainder canceled")
        );
        assert_initializes_order(report);
    }
}