guilder_client_hyperliquid/

client.rs

use crate::rate_limiter::{AddressRateLimiter, RestRateLimiter};
use crate::ws::manager::{
    managed_stream, HyperliquidSubscription, HyperliquidWsManager, WsSendRateLimiter,
};
use crate::ws::{HyperliquidWsBook, HyperliquidWsInboundMessage};
use async_trait::async_trait;
use futures_util::{stream, StreamExt};
use guilder_abstraction::{
    self, AssetContext, BoxStream, Deposit, EcdsaSignature, ExternalSigner, Fill, FundingPayment,
    L2Level, L2Snapshot, L2Update, Liquidation, OpenOrder, OrderPlacement, OrderSide, OrderType,
    OrderUpdate, Position, PredictedFunding, TimeInForce, UserFill, Withdrawal,
};
use reqwest::Client;
use rust_decimal::Decimal;
use serde::Deserialize;
use serde_json::Value;
use std::collections::HashMap;
use std::str::FromStr;
use std::sync::{Arc, RwLock};
const HYPERLIQUID_INFO_URL: &str = "https://api.hyperliquid.xyz/info";
const HYPERLIQUID_EXCHANGE_URL: &str = "https://api.hyperliquid.xyz/exchange";

/// REST/WS endpoints, selectable at construction time. `Mainnet` is the
/// default; `Testnet` points at the official testnet replica
/// (api.hyperliquid-testnet.xyz) — real order-book mechanics, simulated
/// balances (faucet-funded). Configurable per Sho 2026-09-21: cleanest at
/// client construction, not via env or compile-time flags.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum HyperliquidNetwork {
    /// Production: api.hyperliquid.xyz
    #[default]
    Mainnet,
    /// Official testnet replica: api.hyperliquid-testnet.xyz
    Testnet,
}

impl HyperliquidNetwork {
    pub fn info_url(&self) -> &'static str {
        match self {
            Self::Mainnet => HYPERLIQUID_INFO_URL,
            Self::Testnet => "https://api.hyperliquid-testnet.xyz/info",
        }
    }
    pub fn exchange_url(&self) -> &'static str {
        match self {
            Self::Mainnet => HYPERLIQUID_EXCHANGE_URL,
            Self::Testnet => "https://api.hyperliquid-testnet.xyz/exchange",
        }
    }
    pub fn ws_url(&self) -> &'static str {
        match self {
            Self::Mainnet => "wss://api.hyperliquid.xyz/ws",
            Self::Testnet => "wss://api.hyperliquid-testnet.xyz/ws",
        }
    }
    /// EIP-712 phantom-agent `source` for L1 action signing.
    /// Hyperliquid mainnet uses "a", testnet uses "b" — matches the official
    /// Python SDK (`construct_phantom_agent`).
    pub fn eip712_source(&self) -> &'static str {
        match self {
            Self::Mainnet => "a",
            Self::Testnet => "b",
        }
    }
}

async fn parse_response<T: for<'de> serde::Deserialize<'de>>(
    resp: reqwest::Response,
) -> Result<T, String> {
    let status = resp.status();
    let text = resp
        .text()
        .await
        .map_err(|e| format!("failed to read response body (status {status}): {e}"))?;

    if text.is_empty() {
        return Err(format!(
            "empty response body from Hyperliquid (HTTP {status})"
        ));
    }

    serde_json::from_str(&text).map_err(|e| {
        let snippet = if text.len() > 512 {
            format!("{}... ({} bytes total)", &text[..256], text.len())
        } else {
            text.clone()
        };
        format!("deserialize error (HTTP {status}): {e}: {snippet}")
    })
}

pub struct HyperliquidClient {
    client: Client,
    network: HyperliquidNetwork,
    user_address: Option<String>,
    private_key: Option<String>,
    external_signer: Option<Arc<dyn ExternalSigner>>,
    rest_limiter: Arc<RestRateLimiter>,
    address_limiter: Arc<AddressRateLimiter>,
    market_ws_manager: HyperliquidWsManager,
    user_ws_managers: Arc<RwLock<HashMap<String, HyperliquidWsManager>>>,
    ws_send_limiter: WsSendRateLimiter,
    /// Monotonic nonce source (2026-09-28 z4dbg incident): the last nonce this
    /// client handed out. HL rejects a signed action whose nonce was already
    /// used (`Invalid nonce: duplicate nonce N`), and the raw wall-clock
    /// millisecond value repeats whenever two signed actions are built within
    /// the same millisecond — exactly the emergency close fan-out shape.
    /// The provider hands out `max(now_ms, last + 1)` under a compare-and-swap
    /// so concurrent signers can never observe the same nonce twice.
    last_nonce: Arc<std::sync::atomic::AtomicU64>,
    /// T1 asset-index cache (2026-09-30): the universe snapshot rarely moves —
    /// listings are hourly-ish events, not per-order. A fresh w20 `meta` POST
    /// per order/cancel burned ~72% of the REST info budget (the live lane
    /// investigation: submissions AND cancels were rate-limited into ghost
    /// orders). TTL ~10min falls back to refetch when the symbol misses.
    asset_index_cache: Arc<tokio::sync::RwLock<Option<(std::time::Instant, std::collections::HashMap<String, usize>)>>>,
}

impl Default for HyperliquidClient {
    fn default() -> Self {
        Self::new()
    }
}

impl HyperliquidClient {
    pub fn new() -> Self {
        Self::with_network(HyperliquidNetwork::Mainnet)
    }

    /// M2 (Sho 2026-09-21): select the network (mainnet/testnet) at
    /// construction. All REST + WS endpoints follow.
    pub fn with_network(network: HyperliquidNetwork) -> Self {
        let ws_send_limiter = WsSendRateLimiter::new();
        HyperliquidClient {
            client: Client::new(),
            network,
            user_address: None,
            private_key: None,
            external_signer: None,
            rest_limiter: Arc::new(RestRateLimiter::new()),
            address_limiter: Arc::new(AddressRateLimiter::new()),
            market_ws_manager: HyperliquidWsManager::new(
                None,
                ws_send_limiter.clone(),
                network.ws_url(),
            ),
            user_ws_managers: Arc::new(RwLock::new(HashMap::new())),
            ws_send_limiter,
            last_nonce: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            asset_index_cache: Arc::new(tokio::sync::RwLock::new(None)),
        }
    }

    /// Monotonic nonce source (2026-09-28 z4dbg fix): HL rejects a signed
    /// action whose nonce repeats (`Invalid nonce: duplicate nonce N`), and the
    /// raw wall-clock millisecond repeats whenever two actions are signed in
    /// the same millisecond. Returns `max(now_ms, last + 1)` via CAS so
    /// concurrent signers can never observe the same nonce twice.
    pub(crate) fn next_nonce(&self) -> u64 {
        use std::sync::atomic::Ordering;
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        let mut last = self.last_nonce.load(Ordering::Relaxed);
        loop {
            let candidate = now_ms.max(last + 1);
            match self
                .last_nonce
                .compare_exchange(last, candidate, Ordering::AcqRel, Ordering::Relaxed)
            {
                Ok(_) => return candidate,
                Err(observed) => last = observed,
            }
        }
    }

    pub fn with_auth(user_address: impl Into<String>, private_key: String) -> Self {
        Self::with_network_and_auth(HyperliquidNetwork::Mainnet, user_address, private_key)
    }

    pub fn with_network_and_auth(
        network: HyperliquidNetwork,
        user_address: impl Into<String>,
        private_key: String,
    ) -> Self {
        let ws_send_limiter = WsSendRateLimiter::new();
        HyperliquidClient {
            client: Client::new(),
            network,
            user_address: Some(user_address.into()),
            private_key: Some(private_key),
            external_signer: None,
            rest_limiter: Arc::new(RestRateLimiter::new()),
            address_limiter: Arc::new(AddressRateLimiter::new()),
            market_ws_manager: HyperliquidWsManager::new(
                None,
                ws_send_limiter.clone(),
                network.ws_url(),
            ),
            user_ws_managers: Arc::new(RwLock::new(HashMap::new())),
            ws_send_limiter,
            last_nonce: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            asset_index_cache: Arc::new(tokio::sync::RwLock::new(None)),
        }
    }

    /// Create a client authenticated via an external signer (TPM, Secure Enclave, etc.).
    ///
    /// The signer handles the actual ECDSA signing; the private key never leaves
    /// the hardware. The client computes the EIP-712 digest and delegates signing
    /// to the external signer.
    pub fn with_external_signer(
        user_address: impl Into<String>,
        signer: Arc<dyn ExternalSigner>,
    ) -> Self {
        let ws_send_limiter = WsSendRateLimiter::new();
        HyperliquidClient {
            client: Client::new(),
            network: HyperliquidNetwork::Mainnet,
            user_address: Some(user_address.into()),
            private_key: None,
            external_signer: Some(signer),
            rest_limiter: Arc::new(RestRateLimiter::new()),
            address_limiter: Arc::new(AddressRateLimiter::new()),
            market_ws_manager: HyperliquidWsManager::new(
                None,
                ws_send_limiter.clone(),
                HyperliquidNetwork::Mainnet.ws_url(),
            ),
            user_ws_managers: Arc::new(RwLock::new(HashMap::new())),
            ws_send_limiter,
            last_nonce: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            asset_index_cache: Arc::new(tokio::sync::RwLock::new(None)),
        }
    }

    pub fn with_network_and_external_signer(
        network: HyperliquidNetwork,
        user_address: impl Into<String>,
        signer: Arc<dyn ExternalSigner>,
    ) -> Self {
        let ws_send_limiter = WsSendRateLimiter::new();
        HyperliquidClient {
            client: Client::new(),
            network,
            user_address: Some(user_address.into()),
            private_key: None,
            external_signer: Some(signer),
            rest_limiter: Arc::new(RestRateLimiter::new()),
            address_limiter: Arc::new(AddressRateLimiter::new()),
            market_ws_manager: HyperliquidWsManager::new(
                None,
                ws_send_limiter.clone(),
                network.ws_url(),
            ),
            user_ws_managers: Arc::new(RwLock::new(HashMap::new())),
            ws_send_limiter,
            last_nonce: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            asset_index_cache: Arc::new(tokio::sync::RwLock::new(None)),
        }
    }

    /// Configure rate limit budgets (rest_weight/min, address_requests).
    /// Defaults: 1200 rest weight/min, 10000 address requests.
    pub fn with_budgets(mut self, rest_weight: u32, addr_budget: u64) -> Self {
        self.rest_limiter = Arc::new(RestRateLimiter::new_with_budget(rest_weight));
        self.address_limiter = Arc::new(AddressRateLimiter::new_with_budget(addr_budget));
        self
    }

    /// T4 (2026-09-30): snapshot of the REST (info) weight budget —
    /// `(remaining, max, window_entries)`. Read-only; does not consume
    /// budget. Host apps export this as a telemetry gauge so budget
    /// exhaustion (the ghost-order cause for failed cancels) is a
    /// queryable trend instead of a surprise.
    pub async fn rest_budget_snapshot(&self) -> (u32, u32, usize) {
        self.rest_limiter.budget_snapshot().await
    }

    /// POST to the info endpoint, consuming `weight` from the REST rate-limit budget.
    /// Returns `Err("rate_limited: ...")` immediately if budget is exhausted — no retry.
    /// Callers should handle gracefully (skip cycle, retry later, etc.).
    async fn info_post(
        &self,
        body: Value,
        weight: u32,
        call: &str,
    ) -> Result<reqwest::Response, String> {
        self.rest_limiter.acquire(weight).await.map_err(|e| {
            format!(
                "rate_limited: info_post ({call}) budget exhausted, retry_after_ms={}",
                e.retry_after.as_millis()
            )
        })?;
        self.client
            .post(self.network.info_url())
            .json(&body)
            .send()
            .await
            .map_err(|e| e.to_string())
    }
    fn require_user_address(&self) -> Result<String, String> {
        self.user_address
            .clone()
            .ok_or_else(|| "user address required: use HyperliquidClient::with_auth".to_string())
    }

    fn require_private_key(&self) -> Result<Option<&str>, String> {
        if self.external_signer.is_some() {
            return Ok(None);
        }
        self.private_key.as_deref().map(Some).ok_or_else(|| {
            "private key or external signer required: use with_auth or with_external_signer"
                .to_string()
        })
    }

    /// Cache-first asset index lookup (T1, 2026-09-30): serves from the
    /// universe snapshot when fresh; fetches + populates only on miss/TTL.
    /// Saves a w20 `meta` POST per order/cancel after the first call.
    pub(crate) async fn asset_index_lookup(&self, symbol: &str) -> Result<usize, String> {
        const UNIVERSE_TTL: std::time::Duration = std::time::Duration::from_secs(600);
        {
            let cache = self.asset_index_cache.read().await;
            if let Some((at, map)) = cache.as_ref() {
                if at.elapsed() < UNIVERSE_TTL {
                    if let Some(idx) = map.get(symbol) {
                        return Ok(*idx);
                    }
                }
            }
        }
        let resp = self
            .info_post(serde_json::json!({"type": "meta"}), 20, "get_asset_index")
            .await?;
        let meta: MetaResponse = parse_response(resp).await?;
        let map: std::collections::HashMap<String, usize> = meta
            .universe
            .iter()
            .enumerate()
            .map(|(i, a)| (a.name.clone(), i))
            .collect();
        let idx = map
            .get(symbol)
            .copied()
            .ok_or_else(|| format!("symbol {} not found", symbol))?;
        *self.asset_index_cache.write().await = Some((std::time::Instant::now(), map));
        Ok(idx)
    }

    async fn get_asset_index(&self, symbol: &str) -> Result<usize, String> {
        self.asset_index_lookup(symbol).await
    }

    async fn submit_signed_action(
        &self,
        action: Value,
        vault_address: Option<&str>,
    ) -> Result<Value, String> {
        let private_key = self.require_private_key()?;
        let nonce = self.next_nonce();

        let (r, s, v) = sign_action(
            private_key,
            self.external_signer.as_ref(),
            &action,
            vault_address,
            nonce,
            self.network.eip712_source(),
        )
        .await?;

        let payload = serde_json::json!({
            "action": action,
            "nonce": nonce,
            "signature": {"r": r, "s": s, "v": v},
            "vaultAddress": null,
            "expiresAfter": null
        });

        // Check both rate limiters non-blocking — fail fast, no retry.
        self.rest_limiter.acquire(1).await.map_err(|e| {
            format!(
                "rate_limited: rest_weight exhausted, retry_after_ms={}",
                e.retry_after.as_millis()
            )
        })?;
        self.address_limiter.acquire(1, false).await.map_err(|e| {
            format!(
                "rate_limited: address quota exhausted, retry_after_ms={}",
                e.retry_after.as_millis()
            )
        })?;

        let resp = self
            .client
            .post(self.network.exchange_url())
            .json(&payload)
            .send()
            .await
            .map_err(|e| e.to_string())?;

        let status = resp.status();
        if !status.is_success() {
            let text = resp.text().await.map_err(|e| e.to_string())?;
            return Err(format!("HTTP {status}: {text}"));
        }

        let body: Value = parse_response(resp).await?;
        if body["status"].as_str() == Some("err") {
            return Err(body["response"]
                .as_str()
                .unwrap_or("unknown error")
                .to_string());
        }
        Ok(body)
    }
}

// --- REST deserialization types ---

#[derive(Deserialize)]
struct MetaResponse {
    universe: Vec<AssetInfo>,
}

#[derive(Deserialize)]
struct AssetInfo {
    name: String,
    #[serde(rename = "szDecimals")]
    sz_decimals: i32,
    #[serde(rename = "isDelisted", default)]
    is_delisted: bool,
}

type MetaAndAssetCtxsResponse = (MetaResponse, Vec<RestAssetCtx>);

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[allow(dead_code)]
struct RestAssetCtx {
    open_interest: String,
    funding: String,
    mark_px: String,
    day_ntl_vlm: String,
    mid_px: Option<String>,
    oracle_px: Option<String>,
    premium: Option<String>,
    prev_day_px: Option<String>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[allow(dead_code)]
struct ClearinghouseStateResponse {
    margin_summary: MarginSummary,
    asset_positions: Vec<AssetPosition>,
}

/// Kept for get_positions compatibility; margin_summary fields are unused since get_collateral was removed.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
#[allow(dead_code)]
struct MarginSummary {
    account_value: String,
    #[serde(default)]
    total_ntl_pos: Option<String>,
    #[serde(default)]
    total_raw_usd: Option<String>,
    #[serde(default)]
    total_margin_used: Option<String>,
}

#[derive(Deserialize)]
struct AssetPosition {
    position: PositionDetail,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct PositionDetail {
    coin: String,
    /// positive = long, negative = short
    szi: String,
    entry_px: Option<String>,
    /// Venue-authoritative unrealized PnL (HL reports it per position).
    #[serde(default)]
    unrealized_pnl: Option<String>,
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct RestOpenOrder {
    coin: String,
    side: String,
    limit_px: String,
    sz: String,
    oid: i64,
    orig_sz: String,
    cloid: Option<String>,
}

// predictedFundings response: Vec<(coin, Vec<(venue, entry_or_null)>)>
// The API returns null for venues that don't list the coin.
type PredictedFundingsResponse = Vec<(String, Vec<(String, Option<PredictedFundingEntry>)>)>;

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct PredictedFundingEntry {
    funding_rate: String,
    next_funding_time: i64,
}

// --- Helpers ---

/// spotClearinghouseState response. The maintenance map is ABSENT on
/// testnet (field only exists on mainnet) — default = empty map = zero
/// maintenance impact on every token (caught live 2026-09-26: every
/// testnet get_balance failed `missing field` and equity read as 0).
#[derive(Deserialize)]
pub(crate) struct SpotStateResponse {
    pub balances: Vec<SpotBalance>,
    #[serde(default, rename = "tokenToAvailableAfterMaintenance")]
    pub token_to_available_after_maintenance: Vec<(i32, String)>,
}

#[derive(Deserialize)]
pub(crate) struct SpotBalance {
    pub coin: String,
    pub total: String,
    pub hold: String,
    #[serde(default)]
    pub token: Option<i32>,
    #[serde(default)]
    #[serde(rename = "entryNtl")]
    pub entry_ntl: Option<String>,
}

/// Map a parsed SPOT state (spotClearinghouseState) into account balances.
/// Pure — unit-tested against both the mainnet shape (maintenance map
/// present) and the testnet shape (absent).
pub(crate) fn map_spot_state(
    state: SpotStateResponse,
    perp_margin_used: Option<Decimal>,
) -> Result<Vec<guilder_abstraction::AccountBalance>, String> {
    // Build a safe lookup map from token ID → available after maintenance.
    // Not every token appears in the map — absence means zero maintenance impact.
    let safe_map: HashMap<i32, Decimal> = state
        .token_to_available_after_maintenance
        .into_iter()
        .filter_map(|(token_id, value)| parse_decimal(&value).map(|v| (token_id, v)))
        .collect();

    state
        .balances
        .into_iter()
        .map(|balance| {
            let equity =
                parse_decimal(&balance.total).ok_or_else(|| "invalid total balance".to_string())?;
            let hold =
                parse_decimal(&balance.hold).ok_or_else(|| "invalid hold balance".to_string())?;
            let free = equity - hold;

            let safe = balance
                .token
                .and_then(|token_id| safe_map.get(&token_id).copied());
            let usable = match safe {
                Some(s) => std::cmp::min(free, s),
                None => free,
            };
            let maintenance = safe.map(|s| equity - s);

            // margin_used is account-level (perp side); only populate on USDC.
            let margin_used = if balance.coin == "USDC" {
                perp_margin_used
            } else {
                None
            };

            Ok(guilder_abstraction::AccountBalance {
                token: balance.coin,
                equity,
                free,
                safe,
                usable,
                hold,
                margin_used,
                maintenance,
                settled_usd: None,
            })
        })
        .collect()
}

/// Map the PERP margin account (clearinghouseState) into the trading USDC
/// balance row. Under Hyperliquid's manual-account model spot and futures
/// are SEPARATE ledgers — futures trading sizes from THIS account:
/// equity = accountValue, free/usable = accountValue - totalMarginUsed.
pub(crate) fn map_perp_state(
    state: ClearinghouseStateResponse,
) -> Result<guilder_abstraction::AccountBalance, String> {
    let equity = parse_decimal(&state.margin_summary.account_value)
        .ok_or_else(|| "invalid accountValue".to_string())?;
    let margin_used = state
        .margin_summary
        .total_margin_used
        .as_deref()
        .and_then(parse_decimal)
        .unwrap_or(Decimal::ZERO);
    let free = equity - margin_used;
    // settled cash = accountValue − Σ uPnL(assetPositions) — the accounting
    // basis. NOT marginSummary.totalRawUsd: that field is the LIQUIDATION
    // basis (accountValue − notional), not cash (albatross #114 live:
    // anchoring on it produced a −119 equity, delta tracking the notional).
    let venue_uPnL: Decimal = state
        .asset_positions
        .iter()
        .filter_map(|ap| ap.position.unrealized_pnl.as_deref().and_then(parse_decimal))
        .sum();
    let settled_usd = Some(equity - venue_uPnL);
    Ok(guilder_abstraction::AccountBalance {
        token: crate::PERP_LEDGER_TOKEN.to_string(),
        equity,
        free,
        safe: None,
        usable: free,
        hold: Decimal::ZERO,
        margin_used: Some(margin_used),
        maintenance: Some(equity - margin_used),
        settled_usd,
    })
}

fn parse_decimal(s: &str) -> Option<Decimal> {
    Decimal::from_str(s).ok()
}

fn keccak256(data: &[u8]) -> [u8; 32] {
    use sha3::{Digest, Keccak256};
    Keccak256::digest(data).into()
}

/// EIP-712 domain separator for Hyperliquid L1 actions (chainId=1337).
fn hyperliquid_domain_separator() -> [u8; 32] {
    let type_hash = keccak256(
        b"EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)",
    );
    let name_hash = keccak256(b"Exchange");
    let version_hash = keccak256(b"1");
    let mut chain_id = [0u8; 32];
    chain_id[28..32].copy_from_slice(&1337u32.to_be_bytes());
    let verifying_contract = [0u8; 32];

    let mut data = [0u8; 160];
    data[..32].copy_from_slice(&type_hash);
    data[32..64].copy_from_slice(&name_hash);
    data[64..96].copy_from_slice(&version_hash);
    data[96..128].copy_from_slice(&chain_id);
    data[128..160].copy_from_slice(&verifying_contract);
    keccak256(&data)
}

/// Convert a `serde_json::Value` to msgpack bytes, preserving the JSON map key order.
/// This avoids rmp_serde's HashMap-based serialization which reorders map keys.
fn value_to_msgpack(val: &Value) -> Vec<u8> {
    match val {
        Value::Null => vec![0xc0],
        Value::Bool(true) => vec![0xc3],
        Value::Bool(false) => vec![0xc2],
        Value::Number(n) => {
            if let Some(i) = n.as_i64() {
                if i >= 0 {
                    if i <= 127 {
                        vec![i as u8]
                    } else if i <= 255 {
                        vec![0xcc, i as u8]
                    } else if i <= 65535 {
                        let mut buf = vec![0xcd];
                        buf.extend_from_slice(&(i as u16).to_be_bytes());
                        buf
                    } else if i <= 4294967295 {
                        let mut buf = vec![0xce];
                        buf.extend_from_slice(&(i as u32).to_be_bytes());
                        buf
                    } else {
                        let mut buf = vec![0xcf];
                        buf.extend_from_slice(&(i as u64).to_be_bytes());
                        buf
                    }
                } else if i >= -32 {
                    vec![0xe0 | (i as u8)]
                } else if i >= -128 {
                    vec![0xd0, i as i8 as u8]
                } else if i >= -32768 {
                    let mut buf = vec![0xd1];
                    buf.extend_from_slice(&(i as i16).to_be_bytes());
                    buf
                } else if i >= -2147483648 {
                    let mut buf = vec![0xd2];
                    buf.extend_from_slice(&(i as i32).to_be_bytes());
                    buf
                } else {
                    let mut buf = vec![0xd3];
                    buf.extend_from_slice(&i.to_be_bytes());
                    buf
                }
            } else if let Some(f) = n.as_f64() {
                let mut buf = vec![0xcb];
                buf.extend_from_slice(&f.to_be_bytes());
                buf
            } else {
                let u = n.as_u64().unwrap();
                if u <= 127 {
                    vec![u as u8]
                } else if u <= 255 {
                    vec![0xcc, u as u8]
                } else if u <= 65535 {
                    let mut buf = vec![0xcd];
                    buf.extend_from_slice(&(u as u16).to_be_bytes());
                    buf
                } else if u <= 4294967295 {
                    let mut buf = vec![0xce];
                    buf.extend_from_slice(&(u as u32).to_be_bytes());
                    buf
                } else {
                    let mut buf = vec![0xcf];
                    buf.extend_from_slice(&u.to_be_bytes());
                    buf
                }
            }
        }
        Value::String(s) => {
            let bytes = s.as_bytes();
            let len = bytes.len();
            let mut buf = Vec::new();
            if len <= 31 {
                buf.push(0xa0 | (len as u8));
            } else if len <= 255 {
                buf.push(0xd9);
                buf.push(len as u8);
            } else if len <= 65535 {
                buf.push(0xda);
                buf.extend_from_slice(&(len as u16).to_be_bytes());
            } else {
                buf.push(0xdb);
                buf.extend_from_slice(&(len as u32).to_be_bytes());
            }
            buf.extend_from_slice(bytes);
            buf
        }
        Value::Array(arr) => {
            let len = arr.len();
            let mut buf = Vec::new();
            if len <= 15 {
                buf.push(0x90 | (len as u8));
            } else if len <= 65535 {
                buf.push(0xdc);
                buf.extend_from_slice(&(len as u16).to_be_bytes());
            } else {
                buf.push(0xdd);
                buf.extend_from_slice(&(len as u32).to_be_bytes());
            }
            for item in arr {
                buf.extend_from_slice(&value_to_msgpack(item));
            }
            buf
        }
        Value::Object(map) => {
            let len = map.len();
            let mut buf = Vec::new();
            if len <= 15 {
                buf.push(0x80 | (len as u8));
            } else if len <= 65535 {
                buf.push(0xde);
                buf.extend_from_slice(&(len as u16).to_be_bytes());
            } else {
                buf.push(0xdf);
                buf.extend_from_slice(&(len as u32).to_be_bytes());
            }
            for (key, value) in map {
                buf.extend_from_slice(&value_to_msgpack(&Value::String(key.clone())));
                buf.extend_from_slice(&value_to_msgpack(value));
            }
            buf
        }
    }
}

/// Convert action to msgpack bytes preserving JSON map key insertion order
/// (matching Python's msgpack dict ordering).
fn action_to_canonical_msgpack(action: &Value) -> Result<Vec<u8>, String> {
    Ok(value_to_msgpack(action))
}

/// Build msgpack for a single order with Python SDK field order:
/// a, b, p, s, r, t, c(opt)
fn build_order_msgpack(
    asset_idx: usize,
    is_buy: bool,
    price: &str,
    size: &str,
    reduce_only: bool,
    order_kind: &str,
    tif: &[u8],
    cloid: Option<&str>,
) -> Vec<u8> {
    let field_count = if cloid.is_some() { 7 } else { 6 };
    let mut buf = Vec::new();
    buf.push(0x80 | (field_count as u8)); // fixmap

    // "a": asset_idx
    buf.extend_from_slice(&value_to_msgpack(&Value::String("a".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::Number(serde_json::Number::from(
        asset_idx,
    ))));

    // "b": is_buy
    buf.extend_from_slice(&value_to_msgpack(&Value::String("b".to_string())));
    buf.push(if is_buy { 0xc3 } else { 0xc2 });

    // "p": price
    buf.extend_from_slice(&value_to_msgpack(&Value::String("p".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(price.to_string())));

    // "s": size (Python SDK puts s before r)
    buf.extend_from_slice(&value_to_msgpack(&Value::String("s".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(size.to_string())));

    // "r": reduce_only
    buf.extend_from_slice(&value_to_msgpack(&Value::String("r".to_string())));
    buf.push(if reduce_only { 0xc3 } else { 0xc2 });

    // "t": { order_kind: { "tif": tif_str } }
    buf.extend_from_slice(&value_to_msgpack(&Value::String("t".to_string())));
    // Inner: fixmap(1) with order_kind key
    buf.push(0x81);
    buf.extend_from_slice(&value_to_msgpack(&Value::String(order_kind.to_string())));
    // Inner-inner: fixmap(1) with "tif" key
    buf.push(0x81);
    buf.extend_from_slice(&value_to_msgpack(&Value::String("tif".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(
        String::from_utf8_lossy(tif).to_string(),
    )));

    // "c": cloid (optional, appended at END per Python SDK)
    if let Some(c) = cloid {
        buf.extend_from_slice(&value_to_msgpack(&Value::String("c".to_string())));
        buf.extend_from_slice(&value_to_msgpack(&Value::String(c.to_string())));
    }

    buf
}

/// Build msgpack for a trigger order (take profit / stop loss) with Python SDK field order.
/// Trigger orders use: t = { "trigger": { "isMarket": bool, "triggerPx": str, "tpsl": "tp"|"sl" } }
fn build_trigger_order_msgpack(
    asset_idx: usize,
    is_buy: bool,
    price: &str,
    size: &str,
    reduce_only: bool,
    trigger_px: &str,
    is_market: bool,
    tpsl: &str,
    cloid: Option<&str>,
) -> Vec<u8> {
    let field_count = if cloid.is_some() { 7 } else { 6 };
    let mut buf = Vec::new();
    buf.push(0x80 | (field_count as u8)); // fixmap

    // "a": asset_idx
    buf.extend_from_slice(&value_to_msgpack(&Value::String("a".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::Number(serde_json::Number::from(
        asset_idx,
    ))));

    // "b": is_buy
    buf.extend_from_slice(&value_to_msgpack(&Value::String("b".to_string())));
    buf.push(if is_buy { 0xc3 } else { 0xc2 });

    // "p": price (use trigger_px as price for resting, or "0" for market-on-trigger)
    buf.extend_from_slice(&value_to_msgpack(&Value::String("p".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(price.to_string())));

    // "s": size
    buf.extend_from_slice(&value_to_msgpack(&Value::String("s".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(size.to_string())));

    // "r": reduce_only
    buf.extend_from_slice(&value_to_msgpack(&Value::String("r".to_string())));
    buf.push(if reduce_only { 0xc3 } else { 0xc2 });

    // "t": { "trigger": { "isMarket": bool, "triggerPx": str, "tpsl": str } }
    buf.extend_from_slice(&value_to_msgpack(&Value::String("t".to_string())));
    buf.push(0x81); // fixmap(1): "trigger"
    buf.extend_from_slice(&value_to_msgpack(&Value::String("trigger".to_string())));
    // trigger object: fixmap(3)
    buf.push(0x83);
    buf.extend_from_slice(&value_to_msgpack(&Value::String("isMarket".to_string())));
    buf.push(if is_market { 0xc3 } else { 0xc2 });
    buf.extend_from_slice(&value_to_msgpack(&Value::String("triggerPx".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(trigger_px.to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String("tpsl".to_string())));
    buf.extend_from_slice(&value_to_msgpack(&Value::String(tpsl.to_string())));

    // "c": cloid (optional, appended at END)
    if let Some(c) = cloid {
        buf.extend_from_slice(&value_to_msgpack(&Value::String("c".to_string())));
        buf.extend_from_slice(&value_to_msgpack(&Value::String(c.to_string())));
    }

    buf
}

/// Compute the EIP-712 digest for a Hyperliquid exchange action.
///
/// This is the hash that gets signed — extracted so both direct key and
/// external signer paths can share it.
fn compute_eip712_digest(
    msgpack: &[u8],
    nonce: u64,
    vault_address: Option<&str>,
    source: &str,
) -> Result<[u8; 32], String> {
    let mut data = msgpack.to_vec();
    data.extend_from_slice(&nonce.to_be_bytes());
    match vault_address {
        None => data.push(0u8),
        Some(addr) => {
            data.push(1u8);
            let addr_bytes = hex::decode(addr.trim_start_matches("0x"))
                .map_err(|e| format!("invalid vault address: {}", e))?;
            data.extend_from_slice(&addr_bytes);
        }
    }

    let connection_id = keccak256(&data);
    let agent_type_hash = keccak256(b"Agent(string source,bytes32 connectionId)");
    let source_hash = keccak256(source.as_bytes());
    let mut struct_data = [0u8; 96];
    struct_data[..32].copy_from_slice(&agent_type_hash);
    struct_data[32..64].copy_from_slice(&source_hash);
    struct_data[64..96].copy_from_slice(&connection_id);
    let struct_hash = keccak256(&struct_data);

    let domain_sep = hyperliquid_domain_separator();
    let mut final_data = Vec::with_capacity(66);
    final_data.extend_from_slice(b"\x19\x01");
    final_data.extend_from_slice(&domain_sep);
    final_data.extend_from_slice(&struct_hash);
    Ok(keccak256(&final_data))
}

/// Format an EcdsaSignature into the (r, s, v) hex strings expected by Hyperliquid.
fn format_ecdsa_signature(sig: &EcdsaSignature) -> (String, String, u8) {
    let r = format!("0x{}", hex::encode(sig.r));
    let s = format!("0x{}", hex::encode(sig.s));
    let v = 27u8 + sig.v;
    (r, s, v)
}

/// Sign a digest using a raw private key (direct key mode).
fn sign_digest_with_key(
    digest: &[u8; 32],
    private_key: &str,
) -> Result<(String, String, u8), String> {
    use k256::ecdsa::SigningKey;

    let key_bytes = hex::decode(private_key.trim_start_matches("0x"))
        .map_err(|e| format!("invalid private key: {}", e))?;
    let signing_key =
        SigningKey::from_bytes(key_bytes.as_slice().into()).map_err(|e| e.to_string())?;
    let (sig, recovery_id) = signing_key
        .sign_prehash_recoverable(digest)
        .map_err(|e| e.to_string())?;

    let sig_bytes = sig.to_bytes();
    let r = format!("0x{}", hex::encode(&sig_bytes[..32]));
    let s = format!("0x{}", hex::encode(&sig_bytes[32..64]));
    let v = 27u8 + recovery_id.to_byte();

    Ok((r, s, v))
}

/// Sign using pre-built msgpack bytes (bypassing serde_json field ordering).
/// Supports both direct private key and external signer.
async fn sign_with_msgpack(
    msgpack: &[u8],
    private_key: Option<&str>,
    external_signer: Option<&Arc<dyn ExternalSigner>>,
    nonce: u64,
    vault_address: Option<&str>,
    source: &str,
) -> Result<(String, String, u8), String> {
    let digest = compute_eip712_digest(msgpack, nonce, vault_address, source)?;

    if let Some(signer) = external_signer {
        let sig = signer.sign_prehash(&digest).await?;
        return Ok(format_ecdsa_signature(&sig));
    }

    let key = private_key.ok_or_else(|| {
        "no signing method available: provide private_key or external_signer".to_string()
    })?;
    sign_digest_with_key(&digest, key)
}

/// Signs a Hyperliquid exchange action using EIP-712.
/// Returns (r, s, v) where r and s are "0x"-prefixed hex strings and v is 27 or 28.
/// Supports both direct private key and external signer.
async fn sign_action(
    private_key: Option<&str>,
    external_signer: Option<&Arc<dyn ExternalSigner>>,
    action: &Value,
    vault_address: Option<&str>,
    nonce: u64,
    source: &str,
) -> Result<(String, String, u8), String> {
    let msgpack_bytes = action_to_canonical_msgpack(action)?;
    let digest = compute_eip712_digest(&msgpack_bytes, nonce, vault_address, source)?;

    if let Some(signer) = external_signer {
        let sig = signer.sign_prehash(&digest).await?;
        return Ok(format_ecdsa_signature(&sig));
    }

    let key = private_key.ok_or_else(|| {
        "no signing method available: provide private_key or external_signer".to_string()
    })?;
    sign_digest_with_key(&digest, key)
}

// --- Trait implementations ---

#[async_trait]
impl guilder_abstraction::TestServer for HyperliquidClient {
    /// Sends a lightweight allMids request; returns true if the server responds 200 OK.
    async fn ping(&self) -> Result<bool, String> {
        // allMids → weight 2
        self.info_post(serde_json::json!({"type": "allMids"}), 2, "ping")
            .await
            .map(|r| r.status().is_success())
    }

    /// Hyperliquid has no dedicated server-time endpoint; returns local UTC ms.
    async fn get_server_time(&self) -> Result<i64, String> {
        Ok(std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as i64)
            .unwrap_or(0))
    }
}

#[async_trait]
impl guilder_abstraction::GetMarketData for HyperliquidClient {
    /// Returns all perpetual asset names from Hyperliquid's meta endpoint.
    async fn get_symbol(&self) -> Result<Vec<String>, String> {
        // meta → weight 20
        let resp = self
            .info_post(serde_json::json!({"type": "meta"}), 20, "get_symbol")
            .await?;
        parse_response::<MetaResponse>(resp)
            .await
            .map(|r| r.universe.into_iter().map(|a| a.name).collect())
    }

    /// Returns the current open interest for `symbol` from metaAndAssetCtxs.
    async fn get_open_interest(&self, symbol: String) -> Result<Decimal, String> {
        // metaAndAssetCtxs → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "metaAndAssetCtxs"}),
                20,
                "get_open_interest",
            )
            .await?;
        let (meta, ctxs) = parse_response::<Option<MetaAndAssetCtxsResponse>>(resp)
            .await?
            .ok_or_else(|| "metaAndAssetCtxs returned null".to_string())?;
        meta.universe
            .iter()
            .position(|a| a.name == symbol)
            .and_then(|i| ctxs.get(i))
            .and_then(|ctx| parse_decimal(&ctx.open_interest))
            .ok_or_else(|| format!("symbol {} not found", symbol))
    }

    /// Returns a full AssetContext snapshot for `symbol` from metaAndAssetCtxs.
    async fn get_asset_context(&self, symbol: String) -> Result<AssetContext, String> {
        // metaAndAssetCtxs → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "metaAndAssetCtxs"}),
                20,
                "get_asset_context",
            )
            .await?;
        let (meta, ctxs) = parse_response::<Option<MetaAndAssetCtxsResponse>>(resp)
            .await?
            .ok_or_else(|| "metaAndAssetCtxs returned null".to_string())?;
        let idx = meta
            .universe
            .iter()
            .position(|a| a.name == symbol)
            .ok_or_else(|| format!("symbol {} not found", symbol))?;
        let ctx = ctxs
            .get(idx)
            .ok_or_else(|| format!("symbol {} not found", symbol))?;
        Ok(AssetContext {
            symbol,
            open_interest: parse_decimal(&ctx.open_interest).ok_or("invalid open_interest")?,
            funding_rate: parse_decimal(&ctx.funding).ok_or("invalid funding")?,
            mark_price: parse_decimal(&ctx.mark_px).ok_or("invalid mark_px")?,
            day_volume: parse_decimal(&ctx.day_ntl_vlm).ok_or("invalid day_ntl_vlm")?,
            mid_price: ctx.mid_px.as_deref().and_then(parse_decimal),
            oracle_price: ctx.oracle_px.as_deref().and_then(parse_decimal),
            premium: ctx.premium.as_deref().and_then(parse_decimal),
            prev_day_price: ctx.prev_day_px.as_deref().and_then(parse_decimal),
            sz_decimals: meta.universe.get(idx).map(|a| a.sz_decimals).unwrap_or(0),
        })
    }

    /// Fetches metaAndAssetCtxs once and returns all asset contexts in universe order.
    /// Prefer this over repeated `get_asset_context` calls to avoid rate-limiting.
    async fn get_all_asset_contexts(&self) -> Result<Vec<AssetContext>, String> {
        // metaAndAssetCtxs → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "metaAndAssetCtxs"}),
                20,
                "get_all_asset_contexts",
            )
            .await?;
        let (meta, ctxs) = parse_response::<Option<MetaAndAssetCtxsResponse>>(resp)
            .await?
            .ok_or_else(|| "metaAndAssetCtxs returned null".to_string())?;
        let mut result = Vec::with_capacity(meta.universe.len());
        for (asset, ctx) in meta.universe.iter().zip(ctxs.iter()) {
            let Some(open_interest) = parse_decimal(&ctx.open_interest) else {
                continue;
            };
            let Some(funding_rate) = parse_decimal(&ctx.funding) else {
                continue;
            };
            let Some(mark_price) = parse_decimal(&ctx.mark_px) else {
                continue;
            };
            let Some(day_volume) = parse_decimal(&ctx.day_ntl_vlm) else {
                continue;
            };
            result.push(AssetContext {
                symbol: asset.name.clone(),
                open_interest,
                funding_rate,
                mark_price,
                day_volume,
                mid_price: ctx.mid_px.as_deref().and_then(parse_decimal),
                oracle_price: ctx.oracle_px.as_deref().and_then(parse_decimal),
                premium: ctx.premium.as_deref().and_then(parse_decimal),
                prev_day_price: ctx.prev_day_px.as_deref().and_then(parse_decimal),
                sz_decimals: asset.sz_decimals,
            });
        }
        Ok(result)
    }

    /// Returns the number of decimal places for order size for a symbol.
    async fn get_sz_decimals(&self, symbol: String) -> Result<i32, String> {
        let all = self.get_all_sz_decimals().await?;
        all.get(&symbol)
            .copied()
            .ok_or_else(|| format!("symbol {} not found", symbol))
    }

    /// Returns sz_decimals for all symbols from the meta universe.
    async fn get_all_sz_decimals(&self) -> Result<HashMap<String, i32>, String> {
        // metaAndAssetCtxs → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "metaAndAssetCtxs"}),
                20,
                "get_all_sz_decimals",
            )
            .await?;
        let (meta, _) = parse_response::<Option<MetaAndAssetCtxsResponse>>(resp)
            .await?
            .ok_or_else(|| "metaAndAssetCtxs returned null".to_string())?;
        Ok(meta
            .universe
            .into_iter()
            .map(|a| (a.name, a.sz_decimals))
            .collect())
    }

    /// Returns a full L2 orderbook snapshot for `symbol` from the l2Book REST endpoint.
    async fn get_l2_orderbook(&self, symbol: String) -> Result<L2Snapshot, String> {
        // l2Book → weight 2
        let resp = self
            .info_post(
                serde_json::json!({"type": "l2Book", "coin": symbol}),
                2,
                "get_l2_orderbook",
            )
            .await?;
        let book: Option<HyperliquidWsBook> = parse_response(resp).await?;
        let book = match book {
            Some(b) => b,
            None => {
                return Ok(L2Snapshot {
                    symbol,
                    bids: vec![],
                    asks: vec![],
                    sequence: 0,
                })
            }
        };
        let bids = book
            .levels
            .first()
            .into_iter()
            .flatten()
            .filter_map(|level| {
                Some(L2Level {
                    price: parse_decimal(&level.px)?,
                    volume: parse_decimal(&level.sz)?,
                })
            })
            .collect();
        let asks = book
            .levels
            .get(1)
            .into_iter()
            .flatten()
            .filter_map(|level| {
                Some(L2Level {
                    price: parse_decimal(&level.px)?,
                    volume: parse_decimal(&level.sz)?,
                })
            })
            .collect();
        Ok(L2Snapshot {
            symbol: book.coin,
            bids,
            asks,
            sequence: book.time,
        })
    }

    /// Returns the mid-price of `symbol` (e.g. "BTC") from allMids.
    async fn get_price(&self, symbol: String) -> Result<Decimal, String> {
        // allMids → weight 2
        let resp = self
            .info_post(serde_json::json!({"type": "allMids"}), 2, "get_price")
            .await?;
        parse_response::<HashMap<String, String>>(resp)
            .await?
            .get(&symbol)
            .and_then(|s| parse_decimal(s))
            .ok_or_else(|| format!("symbol {} not found", symbol))
    }

    /// Returns predicted funding rates for all symbols across all venues.
    /// Null venue entries (unsupported coins) are silently skipped.
    async fn get_predicted_fundings(&self) -> Result<Vec<PredictedFunding>, String> {
        // predictedFundings → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "predictedFundings"}),
                20,
                "get_predicted_fundings",
            )
            .await?;
        let data: PredictedFundingsResponse = parse_response(resp).await?;
        let mut result = Vec::new();
        for (symbol, venues) in data {
            for (venue, entry) in venues {
                let Some(entry) = entry else { continue };
                if let Some(funding_rate) = parse_decimal(&entry.funding_rate) {
                    result.push(PredictedFunding {
                        symbol: symbol.clone(),
                        venue,
                        funding_rate,
                        next_funding_time_ms: entry.next_funding_time,
                    });
                }
            }
        }
        Ok(result)
    }
}

#[async_trait]
impl guilder_abstraction::ManageOrder for HyperliquidClient {
    /// Places an order on Hyperliquid. Requires `with_auth`. Returns an `OrderPlacement` with
    /// the exchange-assigned order ID. Market orders are submitted as aggressive limit orders (IOC).
    ///
    /// If `cloid` is provided, Hyperliquid attaches it to the order lifecycle — fills and order
    /// updates will carry the same cloid back, enabling end-to-end intent tracing without a
    /// separate order_id mapping.
    ///
    /// Trigger orders (`TakeProfit` / `StopLoss`) require `trigger_price` to be set. The order
    /// activates when the mark price reaches `triggerPx`, then executes as a market or limit
    /// order depending on `time_in_force` (`Ioc` = market, `Gtc` = limit).
    async fn place_order(
        &self,
        symbol: String,
        side: OrderSide,
        price: Decimal,
        volume: Decimal,
        order_type: OrderType,
        time_in_force: TimeInForce,
        trigger_price: Option<Decimal>,
        reduce_only: bool,
        cloid: Option<String>,
    ) -> Result<OrderPlacement, String> {
        // Rate limiting is handled in submit_signed_action (non-blocking).
        let asset_idx = self.get_asset_index(&symbol).await?;
        let is_buy = matches!(side, OrderSide::Buy);

        let tif_str = match time_in_force {
            TimeInForce::Gtc => "Gtc",
            TimeInForce::Ioc => "Ioc",
            TimeInForce::Fok => "Fok",
            TimeInForce::Alo => "Alo",
        };

        let cloid_hex = cloid.clone();

        // Determine if this is a trigger order
        let is_trigger = matches!(order_type, OrderType::TakeProfit | OrderType::StopLoss);

        let (order_msgpack, order_type_json) = if is_trigger {
            // --- Trigger order ---
            let trigger_px = trigger_price
                .ok_or_else(|| format!("{:?} order requires trigger_price to be set", order_type))?
                .normalize()
                .to_string();

            let tpsl = match order_type {
                OrderType::TakeProfit => "tp",
                OrderType::StopLoss => "sl",
                _ => unreachable!(),
            };

            // TimeInForce determines market vs limit on trigger:
            // Ioc = market execution (isMarket: true), Gtc/Alo/Fok = limit (isMarket: false)
            let is_market = matches!(time_in_force, TimeInForce::Ioc);

            // For trigger orders, `p` must be set to the trigger price (not "0"),
            // even for market-on-trigger. Hyperliquid validates this field.
            let price_str = if is_market {
                trigger_px.clone()
            } else {
                price.normalize().to_string()
            };

            let msgpack = build_trigger_order_msgpack(
                asset_idx,
                is_buy,
                &price_str,
                &volume.normalize().to_string(),
                reduce_only,
                &trigger_px,
                is_market,
                tpsl,
                cloid_hex.as_deref(),
            );

            let json_type = if is_market {
                format!(
                    r#"{{"trigger":{{"isMarket":true,"triggerPx":"{}","tpsl":"{}"}}}}"#,
                    trigger_px, tpsl
                )
            } else {
                format!(
                    r#"{{"trigger":{{"isMarket":false,"triggerPx":"{}","tpsl":"{}"}}}}"#,
                    trigger_px, tpsl
                )
            };

            (msgpack, json_type)
        } else {
            // --- Regular limit/market order ---
            let (order_kind, tif_bytes) = match order_type {
                OrderType::Limit => ("limit", tif_str.as_bytes()),
                OrderType::Market => ("limit", b"Ioc".as_slice()),
                _ => unreachable!(),
            };

            let price_str = price.normalize().to_string();
            let size_str = volume.normalize().to_string();

            let msgpack = build_order_msgpack(
                asset_idx,
                is_buy,
                &price_str,
                &size_str,
                reduce_only,
                order_kind,
                tif_bytes,
                cloid_hex.as_deref(),
            );

            let json_type = match order_type {
                OrderType::Limit => format!(r#"{{"limit":{{"tif":"{tif_str}"}}}}"#),
                OrderType::Market => r#"{"limit":{"tif":"Ioc"}}"#.to_string(),
                _ => unreachable!(),
            };

            (msgpack, json_type)
        };

        // Build the action-level msgpack with Python SDK field order (insertion order):
        // type → orders → grouping
        let mut action_msgpack = Vec::new();
        action_msgpack.push(0x83); // fixmap(3)
        action_msgpack.extend_from_slice(&value_to_msgpack(&Value::String("type".to_string())));
        action_msgpack.extend_from_slice(&value_to_msgpack(&Value::String("order".to_string())));
        action_msgpack.extend_from_slice(&value_to_msgpack(&Value::String("orders".to_string())));
        action_msgpack.push(0x91); // fixarray(1)
        action_msgpack.extend_from_slice(&order_msgpack);
        action_msgpack.extend_from_slice(&value_to_msgpack(&Value::String("grouping".to_string())));
        action_msgpack.extend_from_slice(&value_to_msgpack(&Value::String("na".to_string())));

        let cloid_json = if let Some(ref c) = cloid_hex {
            format!(r#","c":"{c}""#)
        } else {
            String::new()
        };

        let reduce_json = if reduce_only { "true" } else { "false" };

        let price_for_json = if is_trigger && matches!(time_in_force, TimeInForce::Ioc) {
            // For market-on-trigger, use trigger price (same as msgpack)
            if let Some(ref tp) = trigger_price {
                tp.normalize().to_string()
            } else {
                price.normalize().to_string()
            }
        } else {
            price.normalize().to_string()
        };

        let action_json_str = format!(
            r#"{{"type":"order","orders":[{{"a":{asset_idx},"b":{is_buy},"p":"{price}","s":"{size}","r":{reduce_json},"t":{order_type_json}{cloid_json}}}],"grouping":"na"}}"#,
            price = price_for_json,
            size = volume.normalize().to_string(),
        );

        // Sign using the canonical msgpack (matching Python's field order)
        let private_key = self.require_private_key()?;
        let nonce = self.next_nonce();

        let (r, s, v) = sign_with_msgpack(
            &action_msgpack,
            private_key,
            self.external_signer.as_ref(),
            nonce,
            None,
            self.network.eip712_source(),
        )
        .await?;

        let payload_str = format!(
            r#"{{"action":{},"nonce":{},"signature":{{"r":"{}","s":"{}","v":{}}},"vaultAddress":null,"expiresAfter":null}}"#,
            action_json_str, nonce, r, s, v
        );

        self.rest_limiter.acquire(1).await.map_err(|e| {
            format!(
                "rate_limited: rest_weight exhausted, retry_after_ms={}",
                e.retry_after.as_millis()
            )
        })?;
        self.address_limiter.acquire(1, false).await.map_err(|e| {
            format!(
                "rate_limited: address quota exhausted, retry_after_ms={}",
                e.retry_after.as_millis()
            )
        })?;

        let resp = self
            .client
            .post(self.network.exchange_url())
            .header("Content-Type", "application/json")
            .body(payload_str)
            .send()
            .await
            .map_err(|e| e.to_string())?;

        let status = resp.status();
        if !status.is_success() {
            let text = resp.text().await.map_err(|e| e.to_string())?;
            return Err(format!("HTTP {status}: {text}"));
        }

        let body: Value = parse_response(resp).await?;
        if body["status"].as_str() == Some("err") {
            return Err(body["response"]
                .as_str()
                .unwrap_or("unknown error")
                .to_string());
        }
        let statuses = &body["response"]["data"]["statuses"][0];

        let (oid, returned_cloid, timestamp_ms) = if let Some(resting) = statuses.get("resting") {
            let oid = resting["oid"]
                .as_i64()
                .ok_or_else(|| format!("resting status missing oid: {}", body))?;
            let returned_cloid = resting["cloid"].as_str().map(|s: &str| s.to_string());
            // resting doesn't include a timestamp
            let ts = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64;
            (oid, returned_cloid, ts)
        } else if let Some(filled) = statuses.get("filled") {
            let oid = filled["oid"]
                .as_i64()
                .ok_or_else(|| format!("filled status missing oid: {}", body))?;
            // filled doesn't include cloid
            let ts = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64;
            (oid, None, ts)
        } else if let Some(error) = statuses.get("error") {
            return Err(error
                .as_str()
                .unwrap_or("order rejected with unknown error")
                .to_string());
        } else {
            return Err(format!("unexpected order status: {}", body));
        };

        Ok(OrderPlacement {
            order_id: oid,
            symbol,
            side,
            price,
            quantity: volume,
            timestamp_ms,
            cloid: returned_cloid.or(cloid),
            order_type,
            trigger_price,
            reduce_only,
        })
    }

    /// Modifies price and size of an existing order by its order ID. Requires `with_auth`.
    /// Fetches the order's current coin and side before submitting the modify action.
    async fn change_order_by_cloid(
        &self,
        cloid: i64,
        price: Decimal,
        volume: Decimal,
    ) -> Result<i64, String> {
        let user = self.require_user_address()?;

        // openOrders → weight 20; get_asset_index → meta weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "openOrders", "user": user}),
                20,
                "change_order_by_cloid",
            )
            .await?;
        let orders: Vec<RestOpenOrder> = parse_response(resp).await?;
        let order = orders
            .iter()
            .find(|o| o.oid == cloid)
            .ok_or_else(|| format!("order {} not found", cloid))?;

        let asset_idx = self.get_asset_index(&order.coin).await?;
        let is_buy = order.side == "B";

        let action = serde_json::json!({
            "type": "batchModify",
            "modifies": [{
                "oid": cloid,
                "order": {
                    "a": asset_idx,
                    "b": is_buy,
                    "p": price.to_string(),
                    "s": volume.to_string(),
                    "r": false,
                    "t": {"limit": {"tif": "Gtc"}}
                }
            }]
        });

        self.submit_signed_action(action, None).await?;
        Ok(cloid)
    }

    /// Cancels a single order by its client order ID (cloid). Requires `with_auth`.
    /// Fetches open orders to resolve the order ID for the matching cloid, then
    /// submits a cancel action using the order ID — this works for all order types
    /// including trigger orders (TakeProfit/StopLoss).
    async fn cancel_order_by_cloid(&self, cloid: String) -> Result<(), String> {
        let user = self.require_user_address()?;

        // openOrders → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "openOrders", "user": user}),
                20,
                "cancel_order_by_cloid",
            )
            .await?;
        let orders: Vec<RestOpenOrder> = parse_response(resp).await?;
        let order = orders
            .iter()
            .find(|o| o.cloid.as_ref() == Some(&cloid))
            .ok_or_else(|| format!("order with cloid {} not found", cloid))?;

        // meta → weight 20
        let meta_resp = self
            .info_post(
                serde_json::json!({"type": "meta"}),
                20,
                "cancel_order_by_cloid",
            )
            .await?;
        let meta: MetaResponse = parse_response(meta_resp).await?;

        let asset_idx = meta
            .universe
            .iter()
            .position(|a| a.name == order.coin)
            .ok_or_else(|| format!("asset {} not found in meta", order.coin))?;

        // Use the same "cancel" action type as cancel_all_order, which is
        // proven to work for all order types including trigger orders.
        let action = serde_json::json!({
            "type": "cancel",
            "cancels": [{"a": asset_idx, "o": order.oid}]
        });

        self.submit_signed_action(action, None).await?;
        Ok(())
    }

    /// Cancels all open orders. Requires `with_auth`.
    /// Fetches all open orders and submits a batch cancel in a single signed request.
    async fn cancel_all_order(&self) -> Result<bool, String> {
        let user = self.require_user_address()?;

        // openOrders → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "openOrders", "user": user}),
                20,
                "cancel_all_order",
            )
            .await?;
        let orders: Vec<RestOpenOrder> = parse_response(resp).await?;
        if orders.is_empty() {
            return Ok(true);
        }

        // meta → weight 20
        let meta_resp = self
            .info_post(serde_json::json!({"type": "meta"}), 20, "cancel_all_order")
            .await?;
        let meta: MetaResponse = parse_response(meta_resp).await?;

        let cancels: Vec<Value> = orders
            .iter()
            .filter_map(|o| {
                let asset_idx = meta.universe.iter().position(|a| a.name == o.coin)?;
                Some(serde_json::json!({"a": asset_idx, "o": o.oid}))
            })
            .collect();

        let action = serde_json::json!({"type": "cancel", "cancels": cancels});
        self.submit_signed_action(action, None).await?;
        Ok(true)
    }
}

#[async_trait]
impl guilder_abstraction::SubscribeMarketData for HyperliquidClient {
    fn subscribe_l2_update(&self, symbol: String) -> BoxStream<Result<L2Update, String>> {
        Box::pin(stream::iter(vec![Err(format!(
            "subscribe_l2_update is unsupported for {symbol}; use subscribe_l2_snapshot"
        ))]))
    }

    fn subscribe_l2_snapshot(&self, symbol: String) -> BoxStream<Result<L2Snapshot, String>> {
        let subscription = HyperliquidSubscription::L2Book { coin: symbol };
        Box::pin(managed_stream(
            self.market_ws_manager.clone(),
            subscription,
            |msg: HyperliquidWsInboundMessage| {
                if let Some(snapshot) = msg.as_l2_snapshot() {
                    vec![Ok(snapshot)]
                } else {
                    vec![]
                }
            },
        ))
    }

    fn subscribe_asset_context(&self, symbol: String) -> BoxStream<Result<AssetContext, String>> {
        let subscription = HyperliquidSubscription::ActiveAssetCtx { coin: symbol };
        Box::pin(managed_stream(
            self.market_ws_manager.clone(),
            subscription,
            |msg: HyperliquidWsInboundMessage| {
                if let Some(ctx) = msg.as_asset_context() {
                    vec![Ok(ctx)]
                } else {
                    vec![]
                }
            },
        ))
    }

    fn subscribe_liquidation(&self, user: String) -> BoxStream<Result<Liquidation, String>> {
        subscribe_user_stream(
            self,
            user.clone(),
            HyperliquidSubscription::UserEvents { user_addr: user },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(liq) = msg.as_liquidation() {
                    vec![Ok(liq)]
                } else {
                    vec![]
                }
            },
        )
    }

    fn subscribe_fill(&self, symbol: String) -> BoxStream<Result<Fill, String>> {
        let subscription = HyperliquidSubscription::Trades { coin: symbol };
        Box::pin(managed_stream(
            self.market_ws_manager.clone(),
            subscription,
            |msg: HyperliquidWsInboundMessage| {
                if let Some(fills) = msg.as_trades() {
                    fills.into_iter().map(Ok).collect()
                } else {
                    vec![]
                }
            },
        ))
    }

    /// Gracefully shut down all market data subscriptions.
    /// Closes all broadcast channels so subscribers exit without reconnecting.
    async fn unsubscribe_all(&self) {
        self.market_ws_manager.shutdown();
    }
}

fn subscribe_user_stream<T, F>(
    client: &HyperliquidClient,
    user_addr: String,
    subscription: HyperliquidSubscription,
    parse: F,
) -> BoxStream<Result<T, String>>
where
    T: Send + 'static,
    F: Fn(HyperliquidWsInboundMessage) -> Vec<Result<T, String>> + Send + Sync + 'static,
{
    let manager = get_or_create_user_manager(
        &client.user_ws_managers,
        client.ws_send_limiter.clone(),
        user_addr,
        client.network.ws_url(),
    );
    Box::pin(async_stream::stream! {
        let stream = managed_stream(manager, subscription, parse);
        tokio::pin!(stream);
        while let Some(item) = stream.next().await {
            yield item;
        }
    })
}

fn get_or_create_user_manager(
    user_ws_managers: &RwLock<HashMap<String, HyperliquidWsManager>>,
    ws_send_limiter: WsSendRateLimiter,
    user_addr: String,
    ws_url: &'static str,
) -> HyperliquidWsManager {
    {
        let managers = user_ws_managers.read().unwrap_or_else(|e| e.into_inner());
        if let Some(manager) = managers.get(&user_addr) {
            return manager.clone();
        }
    }

    let mut managers = user_ws_managers.write().unwrap_or_else(|e| e.into_inner());
    managers
        .entry(user_addr.clone())
        .or_insert_with(|| HyperliquidWsManager::new(Some(user_addr), ws_send_limiter, ws_url))
        .clone()
}

#[async_trait]
impl guilder_abstraction::GetAccountSnapshot for HyperliquidClient {
    /// Returns open positions from `clearinghouseState`. Requires `with_auth`.
    /// Zero-size positions are filtered out. Positive `szi` = long, negative = short.
    async fn get_positions(&self) -> Result<Vec<Position>, String> {
        let user = self.require_user_address()?;
        // clearinghouseState → weight 2
        let resp = self
            .info_post(
                serde_json::json!({"type": "clearinghouseState", "user": user}),
                2,
                "get_positions",
            )
            .await?;
        let state: ClearinghouseStateResponse = parse_response(resp).await?;

        Ok(state
            .asset_positions
            .into_iter()
            .filter_map(|ap| {
                let p = ap.position;
                let size = parse_decimal(&p.szi)?;
                if size.is_zero() {
                    return None;
                }
                // entryPx is null on TRANSIENT reads (position book race) —
                // mapping to 0 poisons downstream uPnL accounting AND the
                // reconcile anchor (albatross #114 live: anchor −149 per
                // BERA notional). Skip positions without a real entry.
                let entry_price = p
                    .entry_px
                    .as_deref()
                    .and_then(parse_decimal);
                let Some(entry_price) = entry_price else {
                    return None;
                };
                let side = if size > Decimal::ZERO {
                    OrderSide::Buy
                } else {
                    OrderSide::Sell
                };
                Some(Position {
                    symbol: p.coin,
                    side,
                    size: size.abs(),
                    entry_price,
                })
            })
            .collect())
    }

    /// Returns resting orders from Hyperliquid's `openOrders` endpoint. Requires `with_auth`.
    /// `filled_quantity` is derived as `origSz - sz` (original size minus remaining size).
    async fn get_open_orders(&self) -> Result<Vec<OpenOrder>, String> {
        let user = self.require_user_address()?;
        // openOrders → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "openOrders", "user": user}),
                20,
                "get_open_orders",
            )
            .await?;
        let orders: Vec<RestOpenOrder> = parse_response(resp).await?;

        Ok(orders
            .into_iter()
            .filter_map(|o| {
                let price = parse_decimal(&o.limit_px)?;
                let quantity = parse_decimal(&o.orig_sz)?;
                let remaining = parse_decimal(&o.sz)?;
                let filled_quantity = quantity - remaining;
                let side = if o.side == "B" {
                    OrderSide::Buy
                } else {
                    OrderSide::Sell
                };
                Some(OpenOrder {
                    order_id: o.oid,
                    symbol: o.coin,
                    side,
                    price,
                    quantity,
                    filled_quantity,
                    order_type: None, // openOrders REST endpoint doesn't return order type
                    trigger_price: None, // trigger info not included in openOrders response
                    reduce_only: false, // default; REST doesn't expose this field
                })
            })
            .collect())
    }

    /// Returns all per-asset balances from `spotClearinghouseState` with margin health.
    /// Also fetches `clearinghouseState` to populate `margin_used` on the USDC entry.
    /// Requires `with_auth`.
    async fn get_balance(&self) -> Result<Vec<guilder_abstraction::AccountBalance>, String> {
        let user = self.require_user_address()?;
        // spotClearinghouseState → weight 15
        let resp = self
            .info_post(
                serde_json::json!({"type": "spotClearinghouseState", "user": user}),
                15,
                "get_balance",
            )
            .await?;

        #[allow(dead_code)]
        #[derive(Deserialize)]
        struct SpotBalance {
            coin: String,
            total: String,
            hold: String,
            #[serde(default)]
            token: Option<i32>,
            #[serde(default)]
            #[serde(rename = "entryNtl")]
            entry_ntl: Option<String>,
        }

        let state: SpotStateResponse = parse_response(resp).await?;

        // Fetch the PERP ledger (clearinghouseState). weight 2.
        let perp_parsed: Option<ClearinghouseStateResponse> = match self
            .info_post(
                serde_json::json!({"type": "clearinghouseState", "user": user}),
                2,
                "get_balance_margin",
            )
            .await
        {
            Ok(resp) => parse_response::<ClearinghouseStateResponse>(resp).await.ok(),
            Err(_) => None,
        };

        // Perp-side margin_used lands on the SPOT USDC row for backwards
        // compatibility (0.6.x consumers).
        let perp_margin_used: Option<Decimal> = perp_parsed.as_ref().and_then(|ch| {
            ch.margin_summary
                .total_margin_used
                .as_deref()
                .and_then(parse_decimal)
        });

        // Spot ledger rows — keep per-asset spot balances (incl. USDC spot
        // cash, which is NOT margin under the manual-account model).
        let mut rows = map_spot_state(state, perp_margin_used)?;

        // Perp ledger row appended LAST (stable position for consumers that
        // index rows): token = PERP_LEDGER_TOKEN. equity = accountValue,
        // free/usable = accountValue − totalMarginUsed, margin fields filled.
        // Failure here is fatal — a trader that cannot read its futures
        // ledger must not fall back to spot sizing.
        let perp_state: ClearinghouseStateResponse = match perp_parsed {
            Some(ch) => ch,
            None => {
                // The margin probe failed — retry once so a transient
                // clearinghouse error cannot silently drop the futures row.
                let resp = self
                    .info_post(
                        serde_json::json!({"type": "clearinghouseState", "user": user}),
                        2,
                        "get_balance_perp_row",
                    )
                    .await?;
                parse_response::<ClearinghouseStateResponse>(resp).await?
            }
        };
        rows.push(map_perp_state(perp_state)?);

        Ok(rows)
    }

    /// Returns the user's address-level API rate limit budget.
    /// Queries Hyperliquid's `userRateLimit` info endpoint for authoritative server-side counts.
    async fn get_user_rate_limit(&self) -> Result<guilder_abstraction::UserRateLimit, String> {
        let user = self.require_user_address()?;
        let resp = self
            .info_post(
                serde_json::json!({"type": "userRateLimit", "user": user}),
                20,
                "get_user_rate_limit",
            )
            .await?;
        let val = parse_response::<Value>(resp).await?;

        let cumulative_volume = val["cumVlm"]
            .as_str()
            .and_then(parse_decimal)
            .ok_or_else(|| "missing or invalid cumVlm".to_string())?;
        let requests_used = val["nRequestsUsed"]
            .as_i64()
            .ok_or_else(|| "missing or invalid nRequestsUsed".to_string())?;
        let requests_cap = val["nRequestsCap"]
            .as_i64()
            .ok_or_else(|| "missing or invalid nRequestsCap".to_string())?;
        let requests_surplus = val["nRequestsSurplus"]
            .as_i64()
            .ok_or_else(|| "missing or invalid nRequestsSurplus".to_string())?;

        Ok(guilder_abstraction::UserRateLimit {
            cumulative_volume,
            requests_used,
            requests_cap,
            requests_surplus,
        })
    }
}

#[async_trait]
impl guilder_abstraction::SubscribeUserEvents for HyperliquidClient {
    fn subscribe_user_fills(&self) -> BoxStream<Result<UserFill, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        subscribe_user_stream(
            self,
            addr.clone(),
            HyperliquidSubscription::UserEvents {
                user_addr: addr.clone(),
            },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(fills) = msg.as_user_fills() {
                    fills.into_iter().map(Ok).collect()
                } else {
                    vec![]
                }
            },
        )
    }

    fn subscribe_order_updates(&self) -> BoxStream<Result<OrderUpdate, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        subscribe_user_stream(
            self,
            addr.clone(),
            HyperliquidSubscription::OrderUpdates {
                user_addr: addr.clone(),
            },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(updates) = msg.as_order_updates() {
                    updates.into_iter().map(Ok).collect()
                } else {
                    vec![]
                }
            },
        )
    }

    fn subscribe_funding_payments(&self) -> BoxStream<Result<FundingPayment, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        subscribe_user_stream(
            self,
            addr.clone(),
            HyperliquidSubscription::UserEvents {
                user_addr: addr.clone(),
            },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(p) = msg.as_funding_payment() {
                    vec![Ok(p)]
                } else {
                    vec![]
                }
            },
        )
    }

    fn subscribe_deposits(&self) -> BoxStream<Result<Deposit, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        subscribe_user_stream(
            self,
            addr.clone(),
            HyperliquidSubscription::NonFundingLedger {
                user_addr: addr.clone(),
            },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(deps) = msg.as_deposits() {
                    deps.into_iter().map(Ok).collect()
                } else {
                    vec![]
                }
            },
        )
    }

    fn subscribe_withdrawals(&self) -> BoxStream<Result<Withdrawal, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        subscribe_user_stream(
            self,
            addr.clone(),
            HyperliquidSubscription::NonFundingLedger {
                user_addr: addr.clone(),
            },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(wds) = msg.as_withdrawals() {
                    wds.into_iter().map(Ok).collect()
                } else {
                    vec![]
                }
            },
        )
    }

    /// Subscribe to spot wallet balance updates for the registered user address.
    fn subscribe_spot_balance(
        &self,
    ) -> BoxStream<Result<Vec<guilder_abstraction::AccountBalance>, String>> {
        let Some(addr) = self.user_address.as_ref() else {
            return Box::pin(stream::iter(vec![Err(
                "user address not registered".to_string()
            )]));
        };
        self.subscribe_spot_balance_with_address(addr.clone())
    }

    /// Subscribe to spot wallet balance updates for a specific address.
    fn subscribe_spot_balance_with_address(
        &self,
        address: String,
    ) -> BoxStream<Result<Vec<guilder_abstraction::AccountBalance>, String>> {
        subscribe_user_stream(
            self,
            address.clone(),
            HyperliquidSubscription::UserEvents { user_addr: address },
            |msg: HyperliquidWsInboundMessage| {
                if let Some(balances) = msg.as_spot_balance() {
                    vec![Ok(balances)]
                } else {
                    vec![]
                }
            },
        )
    }

    async fn unsubscribe_user_events(&self) {
        // Unsubscribe from the market-level user manager if we have a user address.
        if let Some(addr) = &self.user_address {
            self.market_ws_manager.unsubscribe_user(addr);
        }
        // Also unsubscribe from all per-user managers.
        let managers = self
            .user_ws_managers
            .read()
            .unwrap_or_else(|e| e.into_inner());
        for (addr, manager) in managers.iter() {
            manager.unsubscribe_user(addr);
        }
    }
}

#[async_trait]
impl guilder_abstraction::SubscribeMarketDataOps for HyperliquidClient {
    async fn unsubscribe_market_data(&self, symbol: String) {
        self.market_ws_manager.unsubscribe_by_coin(&symbol);
    }
}

#[cfg(test)]
mod msgpack_tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn test_eip712_digest_known_answers() {
        // Reference digests generated with the official hyperliquid-python-sdk
        // (msgpack.packb → action_hash → EIP-712 Agent payload → eth_account
        // sign + ECDSA address-recovery verified). Regen script logic:
        // nonce + action mirrored exactly below.
        // 0.6.4 regression guard: source was hardcoded to "a" (mainnet), which
        // makes every TESTNET signature verify against the wrong phantom agent
        // ("User or API Wallet 0x... does not exist").
        let action = json!({
            "type": "order",
            "orders": [{
                "a": 1, "b": true, "p": "1500.5", "s": "0.05",
                "r": false, "t": {"limit": {"tif": "Gtc"}}
            }],
            "grouping": "na"
        });
        let nonce: u64 = 1758572400123;
        let msgpack = action_to_canonical_msgpack(&action).unwrap();

        let mainnet = compute_eip712_digest(&msgpack, nonce, None, "a").unwrap();
        assert_eq!(
            hex::encode(mainnet),
            "28cd97dd515629633af463bd7edaab14e61b3941b638d410c317cac7e0aed860"
        );

        let testnet = compute_eip712_digest(&msgpack, nonce, None, "b").unwrap();
        assert_eq!(
            hex::encode(testnet),
            "d452e53f806773ca6d7ab5d10147518bff0d8b64a4404dcb062e3d2fb3d85c0b"
        );

        assert_ne!(mainnet, testnet);
        assert_eq!(HyperliquidNetwork::Mainnet.eip712_source(), "a");
        assert_eq!(HyperliquidNetwork::Testnet.eip712_source(), "b");
    }

    #[test]
    fn test_msgpack_null() {
        let result = value_to_msgpack(&Value::Null);
        assert_eq!(result, vec![0xc0]);
    }

    #[test]
    fn test_msgpack_bool() {
        assert_eq!(value_to_msgpack(&Value::Bool(true)), vec![0xc3]);
        assert_eq!(value_to_msgpack(&Value::Bool(false)), vec![0xc2]);
    }

    #[test]
    fn test_msgpack_positive_fixint() {
        // 0–127: positive fixint
        assert_eq!(value_to_msgpack(&json!(0)), vec![0x00]);
        assert_eq!(value_to_msgpack(&json!(1)), vec![0x01]);
        assert_eq!(value_to_msgpack(&json!(127)), vec![0x7f]);
    }

    #[test]
    fn test_msgpack_uint8() {
        // 128–255: uint8
        assert_eq!(value_to_msgpack(&json!(128)), vec![0xcc, 0x80]);
        assert_eq!(value_to_msgpack(&json!(255)), vec![0xcc, 0xff]);
    }

    #[test]
    fn test_msgpack_uint16() {
        // 256–65535: uint16
        assert_eq!(value_to_msgpack(&json!(256)), vec![0xcd, 0x01, 0x00]);
        assert_eq!(value_to_msgpack(&json!(65535)), vec![0xcd, 0xff, 0xff]);
    }

    #[test]
    fn test_msgpack_uint32() {
        // 65536–4294967295: uint32
        assert_eq!(
            value_to_msgpack(&json!(65536)),
            vec![0xce, 0x00, 0x01, 0x00, 0x00]
        );
        assert_eq!(
            value_to_msgpack(&json!(4294967295u64)),
            vec![0xce, 0xff, 0xff, 0xff, 0xff]
        );
    }

    #[test]
    fn test_msgpack_uint64() {
        // >4294967295: uint64
        let big: u64 = 4294967296;
        assert_eq!(
            value_to_msgpack(&json!(big)),
            vec![0xcf, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00]
        );
    }

    #[test]
    fn test_msgpack_negative_fixint() {
        // -1 to -32: negative fixint
        assert_eq!(value_to_msgpack(&json!(-1)), vec![0xff]);
        assert_eq!(value_to_msgpack(&json!(-32)), vec![0xe0]);
    }

    #[test]
    fn test_msgpack_int8() {
        // -33 to -128: int8
        assert_eq!(value_to_msgpack(&json!(-33)), vec![0xd0, 0xdf]);
        assert_eq!(value_to_msgpack(&json!(-128)), vec![0xd0, 0x80]);
    }

    #[test]
    fn test_msgpack_int16() {
        // -129 to -32768: int16
        assert_eq!(value_to_msgpack(&json!(-129)), vec![0xd1, 0xff, 0x7f]);
        assert_eq!(value_to_msgpack(&json!(-32768)), vec![0xd1, 0x80, 0x00]);
    }

    #[test]
    fn test_msgpack_int32() {
        // -32769 to -2147483648: int32
        assert_eq!(
            value_to_msgpack(&json!(-32769)),
            vec![0xd2, 0xff, 0xff, 0x7f, 0xff]
        );
        assert_eq!(
            value_to_msgpack(&json!(-2147483648i64)),
            vec![0xd2, 0x80, 0x00, 0x00, 0x00]
        );
    }

    #[test]
    fn test_msgpack_int64() {
        let val: i64 = -2147483649;
        let result = value_to_msgpack(&json!(val));
        assert_eq!(result[0], 0xd3); // int64 marker
        assert_eq!(result.len(), 9);
    }

    #[test]
    fn test_msgpack_float() {
        let result = value_to_msgpack(&json!(3.14));
        assert_eq!(result[0], 0xcb); // float64 marker
        assert_eq!(result.len(), 9);
    }

    #[test]
    fn test_msgpack_fixstr() {
        // 0–31 bytes: fixstr
        assert_eq!(value_to_msgpack(&json!("")), vec![0xa0]);
        assert_eq!(value_to_msgpack(&json!("hello")), {
            let mut expected = vec![0xa5];
            expected.extend_from_slice(b"hello");
            expected
        });
        let s = "a".repeat(31);
        let result = value_to_msgpack(&json!(s));
        assert_eq!(result[0], 0xbf); // 0xa0 | 31
        assert_eq!(result.len(), 32);
    }

    #[test]
    fn test_msgpack_str8() {
        let s = "a".repeat(32);
        let result = value_to_msgpack(&json!(s));
        assert_eq!(result[0], 0xd9); // str8 marker
        assert_eq!(result[1], 32);
        assert_eq!(result.len(), 34);
    }

    #[test]
    fn test_msgpack_fixarray() {
        // 0–15 elements: fixarray
        assert_eq!(value_to_msgpack(&json!([])), vec![0x90]);
        let result = value_to_msgpack(&json!([1, 2, 3]));
        assert_eq!(result[0], 0x93);
        assert_eq!(result, vec![0x93, 0x01, 0x02, 0x03]);
    }

    #[test]
    fn test_msgpack_fixmap() {
        // 0–15 entries: fixmap
        assert_eq!(value_to_msgpack(&json!({})), vec![0x80]);
        let result = value_to_msgpack(&json!({"a": 1}));
        assert_eq!(result[0], 0x81); // fixmap(1)
        assert_eq!(result, {
            let mut expected = vec![0x81];
            expected.extend_from_slice(&value_to_msgpack(&json!("a")));
            expected.extend_from_slice(&value_to_msgpack(&json!(1)));
            expected
        });
    }

    #[test]
    fn test_msgmap_preserves_insertion_order() {
        // Verify keys are serialized in JSON insertion order, not sorted
        let val = json!({
            "z": 1,
            "a": 2,
            "m": 3
        });
        let result = value_to_msgpack(&val);
        // fixmap(3)
        assert_eq!(result[0], 0x83);
        // First key should be "z" (insertion order), not "a" (sorted)
        assert_eq!(result[1], 0xa1); // fixstr(1)
        assert_eq!(result[2], b'z');
    }

    #[test]
    fn test_msgpack_mixed_array() {
        let val = json!([null, true, false, 42, "hi", [1, 2]]);
        let result = value_to_msgpack(&val);
        assert_eq!(result[0], 0x96); // fixarray(6)
        assert_eq!(result[1], 0xc0); // null
        assert_eq!(result[2], 0xc3); // true
        assert_eq!(result[3], 0xc2); // false
        assert_eq!(result[4], 0x2a); // 42
                                     // "hi" = fixstr(2) + "hi"
        assert_eq!(result[5], 0xa2);
        assert_eq!(result[6], b'h');
        assert_eq!(result[7], b'i');
    }

    #[test]
    fn test_build_order_msgpack_without_cloid() {
        let result = build_order_msgpack(
            0,       // asset index
            true,    // is_buy
            "1000",  // price
            "0.1",   // size
            false,   // reduce_only
            "limit", // order_kind
            b"gtc",  // tif
            None,    // cloid
        );
        // fixmap(6)
        assert_eq!(result[0], 0x86);
    }

    #[test]
    fn test_build_order_msgpack_with_cloid() {
        let result = build_order_msgpack(
            0,                // asset index
            true,             // is_buy
            "1000",           // price
            "0.1",            // size
            false,            // reduce_only
            "limit",          // order_kind
            b"gtc",           // tif
            Some("my-cloid"), // cloid
        );
        // fixmap(7)
        assert_eq!(result[0], 0x87);
    }

    #[test]
    fn test_action_to_canonical_msgpack() {
        let action = json!({
            "type": "order",
            "orders": [{"a": 0, "b": true, "p": "1000", "s": "0.1", "r": false, "t": {"limit": {"tif": "gtc"}}}],
            "grouping": "na"
        });
        let result = action_to_canonical_msgpack(&action).unwrap();
        // fixmap(3)
        assert_eq!(result[0], 0x83);
    }

    #[test]
    fn test_msgpack_matches_rmp_serde_for_simple_values() {
        // Verify our encoding matches rmp_serde for simple scalar values
        use rmp_serde::to_vec;

        for val in [
            json!(0),
            json!(127),
            json!(255),
            json!(1000),
            json!(-1),
            json!(-32),
            json!(-128),
        ] {
            let ours = value_to_msgpack(&val);
            let theirs = to_vec(&val).unwrap();
            assert_eq!(
                ours, theirs,
                "mismatch for {}: ours={:?}, rmp={:?}",
                val, ours, theirs
            );
        }
    }

    #[test]
    fn test_msgpack_string_encoding() {
        use rmp_serde::to_vec;
        for val in [
            json!(""),
            json!("a"),
            json!("hello world"),
            json!("BTC-USD"),
        ] {
            let ours = value_to_msgpack(&val);
            let theirs = to_vec(&val).unwrap();
            assert_eq!(
                ours, theirs,
                "mismatch for {}: ours={:?}, rmp={:?}",
                val, ours, theirs
            );
        }
    }

    #[test]
    fn test_msgpack_bool_encoding() {
        use rmp_serde::to_vec;
        let theirs = to_vec(&json!(true)).unwrap();
        assert_eq!(value_to_msgpack(&json!(true)), theirs);
        let theirs = to_vec(&json!(false)).unwrap();
        assert_eq!(value_to_msgpack(&json!(false)), theirs);
    }

    #[test]
    fn test_msgpack_null_encoding() {
        use rmp_serde::to_vec;
        let theirs = to_vec(&Value::Null).unwrap();
        assert_eq!(value_to_msgpack(&Value::Null), theirs);
    }

    #[test]
    fn test_msgpack_nested_object() {
        let val = json!({
            "outer": {
                "inner": 42
            }
        });
        let result = value_to_msgpack(&val);
        assert_eq!(result[0], 0x81); // fixmap(1)
    }

    #[test]
    fn test_msgpack_empty_containers() {
        assert_eq!(value_to_msgpack(&json!([])), vec![0x90]);
        assert_eq!(value_to_msgpack(&json!({})), vec![0x80]);
    }
}

#[async_trait::async_trait]
impl guilder_abstraction::ListingEventSource for HyperliquidClient {
    /// Authoritative lifecycle events from the venue. Hyperliquid's info API
    /// exposes only the CURRENT universe (+ isDelisted flags) — it has no
    /// historical listing stream, so this returns one synthetic `list` event
    /// per never-delisted symbol (event_time unknown → epoch placeholder) and
    /// a `delist` event per isDelisted symbol. The QDB sync layer upserts
    /// these into token_registry_events; true FIRST-LISTING times for
    /// pre-history coins come from earlier sync runs, not from this call.
    async fn get_listing_events(&self) -> Result<Vec<guilder_abstraction::ListingEvent>, String> {
        // meta → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "meta"}),
                20,
                "get_listing_events",
            )
            .await?;
        let meta = parse_response::<MetaResponse>(resp).await?;
        Ok(meta
            .universe
            .into_iter()
            .map(|a| guilder_abstraction::ListingEvent {
                ticker: a.name,
                exchange: "hyperliquid_perp".to_string(),
                event: if a.is_delisted { "delist" } else { "list" }.to_string(),
                event_time: String::new(),
            })
            .collect())
    }

    async fn get_current_universe(&self) -> Result<Vec<guilder_abstraction::SymbolStatus>, String> {
        // meta → weight 20
        let resp = self
            .info_post(
                serde_json::json!({"type": "meta"}),
                20,
                "get_current_universe",
            )
            .await?;
        let meta = parse_response::<MetaResponse>(resp).await?;
        Ok(meta
            .universe
            .into_iter()
            .map(|a| guilder_abstraction::SymbolStatus {
                ticker: a.name,
                exchange: "hyperliquid_perp".to_string(),
                is_delisted: a.is_delisted,
            })
            .collect())
    }
}

#[cfg(test)]
mod nonce_tests {
    use super::*;

    /// Nonce monotonicity (2026-09-28 z4dbg incident): the order lane emits
    /// signed bursts (emergency close fan-out) where two actions signed in the
    /// same millisecond produced the SAME raw timestamp nonce and HL rejected
    /// the loser with `Invalid nonce: duplicate nonce N`. A shared client must
    /// therefore hand out strictly increasing nonces across threads, even when
    /// the wall clock does not advance between calls.
    #[tokio::test]
    async fn next_nonce_is_strictly_increasing_across_concurrent_calls() {
        let client = HyperliquidClient::new();
        let n1 = client.next_nonce();
        let n2 = client.next_nonce();
        let n3 = client.next_nonce();
        assert!(n2 > n1, "nonce must strictly increase: {n1} -> {n2}");
        assert!(n3 > n2, "nonce must strictly increase: {n2} -> {n3}");
    }

    /// The nonce must track the wall clock (HL rejects stale nonces far in the
    /// past) while never reusing or going backwards when the clock stalls
    /// between two consecutive calls: the floor is last_nonce + 1.
    #[tokio::test]
    async fn next_nonce_tracks_clock_but_never_repeats_or_regresses() {
        let client = HyperliquidClient::new();
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64;
        let n1 = client.next_nonce();
        assert!(
            n1 >= now_ms,
            "first nonce must track the current time: {n1} < {now_ms}"
        );
        let n2 = client.next_nonce();
        assert!(n2 > n1, "even a stalled clock must yield n > last ({n1})");
    }

    /// Concurrent signed bursts (the emergency fan-out shape): N threads each
    /// take one nonce and all N must be distinct AND strictly ordered by
    /// acquisition — the exact scenario that produced the duplicate.
    #[tokio::test]
    async fn concurrent_nonce_take_is_duplicate_free() {
        let client = Arc::new(HyperliquidClient::new());
        let handles: Vec<_> = (0..8)
            .map(|_| {
                let c = Arc::clone(&client);
                std::thread::spawn(move || c.next_nonce())
            })
            .collect();
        let mut taken: Vec<u64> = handles.into_iter().map(|h| h.join().unwrap()).collect();
        taken.sort();
        let len = taken.len();
        taken.dedup();
        assert_eq!(
            taken.len(),
            len,
            "concurrent takes must never produce a duplicate nonce: {taken:?}"
        );
    }
}

#[cfg(test)]
mod spot_state_tests {
    use super::*;

    /// EXACT response shape Hyperliquid TESTNET returns for
    /// spotClearinghouseState — NO tokenToAvailableAfterMaintenance.
    /// (Captured live from trader.daometric.com's deserialize warning,
    /// 2026-09-26.) Must parse: USDC total 969, free 969, usable 969.
    const TESTNET_SHAPE: &str =
        r#"{"balances":[{"coin":"USDC","token":0,"total":"969.0","hold":"0.0","entryNtl":"0.0"}]}"#;

    /// MAINNET shape — maintenance map present.
    const MAINNET_SHAPE: &str = r#"{"balances":[{"coin":"USDC","token":0,"total":"969.0","hold":"10.0","entryNtl":"0.0"}],"tokenToAvailableAfterMaintenance":[[0,"955.0"]]}"#;

    #[test]
    fn testnet_spot_state_without_maintenance_map_parses() {
        let state: SpotStateResponse =
            serde_json::from_str(TESTNET_SHAPE).expect("testnet shape must deserialize");
        let balances = map_spot_state(state, None).expect("mapping must succeed");
        assert_eq!(balances.len(), 1);
        let usdc = &balances[0];
        assert_eq!(usdc.token, "USDC");
        assert_eq!(usdc.equity.to_string(), "969.0");
        assert_eq!(usdc.hold.to_string(), "0.0");
        assert_eq!(usdc.free.to_string(), "969.0");
        // No maintenance map → usable falls back to free.
        assert_eq!(usdc.usable.to_string(), "969.0");
        assert_eq!(usdc.safe, None);
        assert_eq!(usdc.maintenance, None);
    }

    #[test]
    fn mainnet_spot_state_with_maintenance_map_parses() {
        let state: SpotStateResponse =
            serde_json::from_str(MAINNET_SHAPE).expect("mainnet shape must deserialize");
        let balances = map_spot_state(state, None).expect("mapping must succeed");
        let usdc = &balances[0];
        assert_eq!(usdc.equity.to_string(), "969.0");
        assert_eq!(usdc.free.to_string(), "959.0");
        // maintenance map present → usable = min(free, safe) = 955.
        assert_eq!(usdc.usable.to_string(), "955.0");
        assert_eq!(usdc.safe.map(|d| d.to_string()), Some("955.0".to_string()));
        assert_eq!(
            usdc.maintenance.map(|d| d.to_string()),
            Some("14.0".to_string())
        );
    }

    #[test]
    fn usdc_balance_carries_perp_margin_used() {
        let state: SpotStateResponse =
            serde_json::from_str(TESTNET_SHAPE).expect("testnet shape must deserialize");
        let balances = map_spot_state(state, Some(Decimal::from(2))).expect("mapping must succeed");
        let usdc = &balances[0];
        assert_eq!(
            usdc.margin_used.map(|d| d.to_string()),
            Some("2".to_string())
        );
    }

    /// 0.7.0 contract: `map_balance_rows` (spot + perp) appends the futures
    /// ledger row LAST with the marker token. Locked by the trader's
    /// ledger-split consumer (account manager).
    #[test]
    fn balance_rows_end_with_marked_perp_row() {
        let state: SpotStateResponse =
            serde_json::from_str(TESTNET_SHAPE).expect("testnet shape must deserialize");
        let perp: ClearinghouseStateResponse = serde_json::from_str(
            r#"{"marginSummary":{"accountValue":"30.0","totalMarginUsed":"5.0"},"assetPositions":[{"position":{"coin":"BTC","szi":"0.1","entryPx":"100","unrealizedPnl":"2.0"}}]}"#,
        )
        .expect("perp shape must deserialize");
        let mut rows =
            map_spot_state(state, Some(Decimal::from(2))).expect("spot rows must map");
        rows.push(map_perp_state(perp).expect("perp row must map"));

        assert_eq!(rows.len(), 2);
        assert_eq!(rows[0].token, "USDC");
        assert_eq!(rows[0].margin_used.map(|d| d.to_string()), Some("2".to_string()));
        assert_eq!(rows[1].token, crate::PERP_LEDGER_TOKEN);
        assert_eq!(rows[1].equity.to_string(), "30.0");
        assert_eq!(rows[1].free.to_string(), "25.0");
        assert_eq!(rows[1].usable.to_string(), "25.0");
        assert_eq!(
            rows[1].margin_used.map(|d| d.to_string()),
            Some("5.0".to_string())
        );
        // 0.7.4: settled cash = accountValue − Σ venue uPnL = 30 − 2 = 28
        // (NOT totalRawUsd — that's the liquidation basis, ≠ cash)
        assert_eq!(rows[1].settled_usd.map(|d| d.to_string()), Some("28.0".into()));
        // Marker token must never collide with a real asset symbol.
        assert!(rows[..rows.len() - 1]
            .iter()
            .all(|b| b.token != crate::PERP_LEDGER_TOKEN));
    }

    /// PERP margin account (clearinghouseState) — the trading money under
    /// Hyperliquid's manual-account model (spot and futures are SEPARATE
    /// ledgers; SMR trades perps, so sizing reads THIS account).
    #[test]
    fn perp_margin_account_maps_to_usdc_row() {
        let state: ClearinghouseStateResponse = serde_json::from_str(
            r#"{"marginSummary":{"accountValue":"30.0","totalNtlPos":"0.0","totalRawUsd":"-70.0","totalMarginUsed":"0.0"},"assetPositions":[]}"#,
        )
        .expect("perp shape must deserialize");
        let usdc = map_perp_state(state).expect("perp mapping must succeed");
        // 0.7.0: the perp ledger row is marked, not "USDC" — spot and futures
        // are separate ledgers and consumers split rows by token.
        assert_eq!(usdc.token, crate::PERP_LEDGER_TOKEN);
        // 0.7.2: settled_usd carries totalRawUsd (venue settled cash) when present.
        // settled = accountValue − ΣuPnL = 30 − 0 = 30 (totalRawUsd −70 ignored)
        assert_eq!(usdc.settled_usd.map(|d| d.to_string()), Some("30.0".into()));
        assert_eq!(usdc.equity.to_string(), "30.0"); // accountValue
        assert_eq!(usdc.free.to_string(), "30.0"); // accountValue - marginUsed
        assert_eq!(usdc.usable.to_string(), "30.0");
        assert_eq!(
            usdc.margin_used.map(|d| d.to_string()),
            Some("0.0".to_string())
        );
    }
}

// T1 asset-index cache (2026-09-30): a w20 `meta` POST per order/cancel
// burned ~72% of the REST info budget (live-lane investigation). Contracts:
// 1) same symbol twice within TTL = ONE network fetch;
// 2) unknown symbol falls back to refetch and still errors;
// 3) cache is per-network (mainnet/testnet never share).
#[cfg(test)]
mod asset_index_cache_tests {
    use super::*;

    #[tokio::test]
    async fn cache_hits_skip_second_fetch() {
        // with a live server this would hit the wire; this test pins the
        // CONTRACT at the cache layer: after populate, get_or_fetch serves
        // from cache without consuming budget.
        let client = HyperliquidClient::new();
        {
            let mut cache = client.asset_index_cache.write().await;
            let mut map = std::collections::HashMap::new();
            map.insert("BTC".to_string(), 0usize);
            *cache = Some((std::time::Instant::now(), map));
        }
        // The lookup path must serve from cache: assert via the same
        // helper `get_asset_index` uses.
        let idx = client.asset_index_lookup("BTC").await.expect("hit");
        assert_eq!(idx, 0, "cached universe serves the symbol");
    }
}