wickra-exchange-core 0.2.1

//! HTX (formerly Huobi) — the seventh exchange.
//!
//! HTX signs AWS-style: the signature is
//! `base64(HMAC-SHA256(secret, METHOD\nhost\npath\nsorted-urlencoded-params))`,
//! where the params include `AccessKeyId`, `SignatureMethod=HmacSHA256`,
//! `SignatureVersion=2` and an ISO-8601 (no-millis) `Timestamp`, and the
//! signature is appended to the query as `Signature=`. Symbols are lowercase and
//! concatenated (`btcusdt`); market-data JSON encodes numbers as JSON numbers
//! (not strings); orders require a spot `account-id` (fetched and cached). The
//! response envelope carries `status: "ok"` with the payload under `tick` or
//! `data`.
//!
//! When built with a futures [`MarketType`](crate::MarketType), the client
//! targets the USDT-margined swap host (`api.hbdm.com`): market data via
//! `/linear-swap-ex/market/*` (same tick/data shapes as spot, so the parsers are
//! reused), and the [`Derivatives`] trait plus `place_order`/`balances` via the
//! **cross-margin** `/linear-swap-api/v1/swap_cross_*` family, where orders carry
//! an integer contract `volume`, a `direction`, an `offset` (open/close) and a
//! `lever_rate`. `query_order`/`cancel_order`/`open_orders` route to the
//! cross-swap order endpoints (`swap_cross_order_info` / `swap_cross_cancel` /
//! `swap_cross_openorders`) with the swap order shape (`order_id_str`,
//! `direction`, numeric `status`). `set_margin_mode(Isolated)` is unsupported
//! within the cross family — a documented gap.
//!
//! [`AdvancedOrders`]: native spot batch place/cancel (`/v1/order/batch-orders`,
//! `/v1/order/orders/batchcancel`, per-order `err-code`). HTX has no STP field,
//! no in-place amend and no OCO order-list, so `amend_order`/`place_oco` and STP
//! are documented gaps.

use crate::clock::ServerClock;
use crate::credentials::Credentials;
use crate::error::{Error, Result};
use crate::events::{BookDelta, BookLevel, Event, OrderBookSnapshot, TradePrint};
use crate::feeds::{
    DerivativesChannel, DerivativesFeed, FundingRate, Liquidation, LongShortRatio, OpenInterest,
};
use crate::idempotency::ClientIdGenerator;
use crate::normalize::{format_decimal, parse_decimal};
use crate::options::{ExchangeOptions, MarginMode, MarketType, SelfTradePrevention};
use crate::positions::{Position, PositionSide};
use crate::signing::hmac_sha256_base64;
use crate::symbol::Symbol;
use crate::traits::{
    AdvancedOrders, Derivatives, DerivativesStream, Exchange, Execution, MarketData, WsExecution,
    WsUserData,
};
use crate::transport::{HttpMethod, HttpRequest, HttpTransport, WsConnection, WsTransport};
use crate::types::{
    Balance, OcoRequest, Order, OrderRequest, OrderSide, OrderStatus, OrderType, Ticker,
    TimeInForce,
};
use rust_decimal::prelude::ToPrimitive;
use rust_decimal::Decimal;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::fmt;
use wickra_core::Candle;

/// Spot API host.
const HOST: &str = "api.huobi.pro";
/// USDT-margined swap (futures) API host.
const FUTURES_HOST: &str = "api.hbdm.com";

/// HTX's public notification socket: funding and forced orders, no
/// credentials. Its market data comes from a different host, so the two are
/// separate connections.
const NOTIFICATION_WS_URL: &str = "wss://api.hbdm.com/linear-swap-notification";

fn system_now_ms() -> i64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .expect("system clock is before the Unix epoch")
        .as_millis() as i64
}

/// Unix milliseconds to an ISO-8601 UTC timestamp without milliseconds
/// (`YYYY-MM-DDTHH:MM:SS`), the form HTX signs.
fn iso8601_no_millis(ms: i64) -> String {
    let days = ms.div_euclid(86_400_000);
    let mut rem = ms.rem_euclid(86_400_000) / 1000;
    let secs = rem % 60;
    rem /= 60;
    let mins = rem % 60;
    let hours = rem / 60;
    let (year, month, day) = civil_from_days(days);
    format!("{year:04}-{month:02}-{day:02}T{hours:02}:{mins:02}:{secs:02}")
}

fn civil_from_days(days: i64) -> (i64, i64, i64) {
    let z = days + 719_468;
    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
    let doe = z - era * 146_097;
    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
    let year = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let day = doy - (153 * mp + 2) / 5 + 1;
    let month = if mp < 10 { mp + 3 } else { mp - 9 };
    (if month <= 2 { year + 1 } else { year }, month, day)
}

/// Percent-encode per RFC 3986 (unreserved characters pass through).
fn encode(s: &str) -> String {
    const HEX: &[u8; 16] = b"0123456789ABCDEF";
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(char::from(b));
            }
            _ => {
                out.push('%');
                out.push(char::from(HEX[(b >> 4) as usize]));
                out.push(char::from(HEX[(b & 0x0f) as usize]));
            }
        }
    }
    out
}

/// An HTX client over injected transports.
pub struct Htx {
    http: Box<dyn HttpTransport>,
    ws: Option<Box<dyn WsTransport>>,
    rest_base: String,
    host: &'static str,
    market_type: MarketType,
    credentials: Option<Credentials>,
    now_ms: Box<dyn Fn() -> i64 + Send + Sync>,
    /// Client order ids for orders the caller did not name.
    ///
    /// A spot algo order is addressed by its client order id -- that is how it
    /// is cancelled -- so one is always sent, seeded from the clock and
    /// monotonic from there.
    client_ids: ClientIdGenerator,
    /// Offset between this machine's clock and the venue's, applied to every
    /// signed timestamp. Zero until [`sync_time`](Self::sync_time) is called.
    clock: ServerClock,
    connection: Option<Box<dyn WsConnection>>,
    sub_messages: Vec<String>,
    subscriptions: Vec<(String, Symbol)>,
    account_id: RefCell<Option<String>>,
    /// Leverage applied to futures orders (HTX sets `lever_rate` per order).
    leverage: Cell<u32>,
    /// The private user-data connection, opened by
    /// [`subscribe_user_data`](Self::subscribe_user_data) and drained by
    /// [`poll_events`](Self::poll_events) alongside the public stream.
    private_connection: Option<Box<dyn WsConnection>>,
    /// The public notification socket, which carries funding and forced
    /// orders. HTX serves those from a different host than its market data,
    /// so this is a third connection rather than another topic on the first.
    notification_connection: Option<Box<dyn WsConnection>>,
    /// Which derivatives channels are subscribed, per contract code.
    derivatives_channels: Vec<(String, DerivativesChannel)>,
    /// Set once the private stream is subscribed, so [`poll_events`](Self::poll_events)
    /// re-subscribes it after a drop.
    user_data_active: bool,
}

/// Hand-written: the client holds `Box<dyn HttpTransport>`, `Box<dyn WsTransport>`
/// and a `Box<dyn Fn() -> i64>` clock, none of which a derive can reach. The
/// transports are shown by whether a connection is open rather than by value, and
/// the credentials only by whether they are set -- printing them would put secret
/// material into every log line that formats a client.
impl fmt::Debug for Htx {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Htx")
            .field("ws", &self.ws.is_some())
            .field("rest_base", &self.rest_base)
            .field("host", &self.host)
            .field("market_type", &self.market_type)
            .field("authenticated", &self.credentials.is_some())
            .field("clock_offset_ms", &self.clock.offset_ms())
            .field("connection", &self.connection.is_some())
            .field("sub_messages", &self.sub_messages.len())
            .field("subscriptions", &self.subscriptions.len())
            .field("account_id", &self.account_id)
            .field("leverage", &self.leverage)
            .field("private_connection", &self.private_connection.is_some())
            .field("user_data_active", &self.user_data_active)
            .finish_non_exhaustive()
    }
}

impl Htx {
    /// Refuse a market this client does not route.
    ///
    /// Called at every seam that reaches the venue -- the HTTP helpers and each
    /// WebSocket connect -- so an unrouted market is refused before a request
    /// is built rather than answered by whichever market the URL happened to
    /// name. See [`ensure_market_is_routed`](super::ensure_market_is_routed)
    /// for what each client used to answer instead.
    fn ensure_market_is_routed(&self) -> Result<()> {
        super::ensure_market_is_routed("HTX", self.market_type, super::SPOT_AND_LINEAR)
    }

    fn build(
        http: Box<dyn HttpTransport>,
        options: &ExchangeOptions,
        credentials: Option<Credentials>,
    ) -> Self {
        let host = if options.market_type.is_derivatives() {
            FUTURES_HOST
        } else {
            HOST
        };
        Self {
            http,
            ws: None,
            rest_base: format!("https://{host}"),
            host,
            market_type: options.market_type,
            credentials,
            now_ms: Box::new(system_now_ms),
            client_ids: ClientIdGenerator::with_seed(
                "wkalgo",
                u64::try_from(system_now_ms()).unwrap_or(0),
            ),
            clock: ServerClock::new(),
            connection: None,
            sub_messages: Vec::new(),
            subscriptions: Vec::new(),
            account_id: RefCell::new(None),
            leverage: Cell::new(1),
            private_connection: None,
            notification_connection: None,
            derivatives_channels: Vec::new(),
            user_data_active: false,
        }
    }

    /// The timestamp a signed request must carry: this machine's clock plus the
    /// offset learned from the venue.
    ///
    /// A venue rejects a signed request whose timestamp falls outside its own
    /// receive window, so a machine a few seconds off has every order refused --
    /// with a message about the window rather than about the clock. Until
    /// [`sync_time`](Self::sync_time) is called the offset is zero and this is
    /// the local time, which is the previous behaviour.
    fn signed_now_ms(&self) -> i64 {
        self.clock.server_time_ms((self.now_ms)())
    }

    /// Learn the offset between this machine's clock and the venue's, from
    /// `GET /v1/common/timestamp`.
    ///
    /// Explicit rather than automatic: it costs a request, and a client should
    /// not make one the caller did not ask for. Call it once after connecting,
    /// and again if the process runs long enough for drift to matter.
    ///
    /// Returns the new offset in milliseconds (`server - local`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the response cannot be parsed.
    pub fn sync_time(&mut self) -> Result<i64> {
        let local_before = (self.now_ms)();
        let value = self.get("/v1/common/timestamp", "")?;
        let server_ms = value
            .get("data")
            .and_then(serde_json::Value::as_i64)
            .ok_or_else(|| Error::Deserialization("common/timestamp: no data".into()))?;
        self.clock.sync(local_before, server_ms);
        Ok(self.clock.offset_ms())
    }

    /// Whether this client targets HTX USDT-margined swaps (`api.hbdm.com`,
    /// `/linear-swap-*`) rather than spot.
    fn is_futures(&self) -> bool {
        self.market_type.is_derivatives()
    }

    /// The HTX **swap** contract code for a canonical [`Symbol`]
    /// (`BTC/USDT` -> `BTC-USDT`, uppercase dash form).
    fn contract_code(symbol: &Symbol) -> String {
        format!(
            "{}-{}",
            symbol.base().to_uppercase(),
            symbol.quote().to_uppercase()
        )
    }

    /// Build a public HTX client.
    #[must_use]
    pub fn with_http(http: Box<dyn HttpTransport>, options: &ExchangeOptions) -> Self {
        Self::build(http, options, None)
    }

    /// Build an authenticated HTX client.
    #[must_use]
    pub fn with_credentials(
        http: Box<dyn HttpTransport>,
        options: &ExchangeOptions,
        credentials: Credentials,
    ) -> Self {
        Self::build(http, options, Some(credentials))
    }

    /// Override the timestamp source (deterministic signing in tests).
    #[must_use]
    pub fn with_clock(mut self, now_ms: Box<dyn Fn() -> i64 + Send + Sync>) -> Self {
        self.now_ms = now_ms;
        self
    }

    /// Attach a WebSocket transport.
    #[must_use]
    pub fn with_ws(mut self, ws: Box<dyn WsTransport>) -> Self {
        self.ws = Some(ws);
        self
    }

    /// The HTX wire symbol for a canonical [`Symbol`] (`BTC/USDT` -> `btcusdt`).
    #[must_use]
    pub fn wire_symbol(symbol: &Symbol) -> String {
        symbol.to_concatenated().to_lowercase()
    }

    /// A ticker for `symbol`.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the symbol is unknown.
    pub fn ticker(&self, symbol: &Symbol) -> Result<Ticker> {
        let (path, query) = if self.is_futures() {
            (
                "/linear-swap-ex/market/detail/merged",
                format!("contract_code={}", Self::contract_code(symbol)),
            )
        } else {
            (
                "/market/detail/merged",
                format!("symbol={}", Self::wire_symbol(symbol)),
            )
        };
        let value = self.get(path, &query)?;
        let tick = value
            .get("tick")
            .ok_or_else(|| Error::Deserialization("missing tick".to_string()))?;
        Ok(Ticker {
            symbol: symbol.clone(),
            last: decimal_field(tick, "close")?,
            bid: decimal_at(tick, "bid", 0)?,
            ask: decimal_at(tick, "ask", 0)?,
            volume: decimal_field(tick, "vol")?,
            timestamp: value
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })
    }

    /// Up to `limit` candles for `symbol` at `interval` (unified). HTX returns
    /// newest-first; the result is chronological.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or a row cannot be parsed.
    pub fn klines(&self, symbol: &Symbol, interval: &str, limit: u32) -> Result<Vec<Candle>> {
        let (path, query) = if self.is_futures() {
            (
                "/linear-swap-ex/market/history/kline",
                format!(
                    "contract_code={}&period={}&size={limit}",
                    Self::contract_code(symbol),
                    map_period(interval),
                ),
            )
        } else {
            (
                "/market/history/kline",
                format!(
                    "symbol={}&period={}&size={limit}",
                    Self::wire_symbol(symbol),
                    map_period(interval),
                ),
            )
        };
        let value = self.get(path, &query)?;
        let data = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing kline data".to_string()))?;
        let mut candles = data
            .iter()
            .map(parse_kline_obj)
            .collect::<Result<Vec<_>>>()?;
        candles.reverse();
        Ok(candles)
    }

    /// A depth snapshot of `symbol` up to `depth` levels per side.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the response cannot be parsed.
    pub fn order_book(&self, symbol: &Symbol, _depth: u32) -> Result<OrderBookSnapshot> {
        let (path, query) = if self.is_futures() {
            (
                "/linear-swap-ex/market/depth",
                format!("contract_code={}&type=step0", Self::contract_code(symbol)),
            )
        } else {
            (
                "/market/depth",
                format!("symbol={}&type=step0", Self::wire_symbol(symbol)),
            )
        };
        let value = self.get(path, &query)?;
        let tick = value
            .get("tick")
            .ok_or_else(|| Error::Deserialization("missing tick".to_string()))?;
        Ok(OrderBookSnapshot {
            symbol: symbol.clone(),
            last_update_id: tick
                .get("version")
                .and_then(serde_json::Value::as_u64)
                .unwrap_or(0),
            bids: num_levels(tick.get("bids"))?,
            asks: num_levels(tick.get("asks"))?,
            timestamp: tick
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })
    }

    /// Subscribe to a pushed derivatives channel.
    ///
    /// HTX serves these from `linear-swap-notification`, a different host from
    /// its market data, so this opens a second public socket. Both topics are
    /// anonymous -- no credentials.
    ///
    /// The funding frame carries no price, so a funding print reports a mark of
    /// zero rather than one taken from another channel at another moment.
    ///
    /// `MarkIndex` is refused: HTX publishes the mark price and the index as
    /// separate kline channels, never in one frame, and pairing them would
    /// present two readings from different moments as one.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, for `MarkIndex`, or if the
    /// subscription fails.
    pub fn subscribe_derivatives(
        &mut self,
        symbol: &Symbol,
        channel: DerivativesChannel,
    ) -> Result<()> {
        self.ensure_market_is_routed()?;
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "HTX",
                "a derivatives channel on a spot client",
                "funding and forced orders exist only on the swap venue",
            ));
        }
        let topic = match channel {
            DerivativesChannel::Funding => "funding_rate",
            DerivativesChannel::Liquidations => "liquidation_orders",
            DerivativesChannel::MarkIndex => {
                return Err(Error::unsupported_field(
                    "HTX",
                    "a combined mark/index channel",
                    "HTX publishes mark and index as separate kline channels, never \
                     in one frame",
                ))
            }
        };
        let code = Self::contract_code(symbol);
        if self.notification_connection.is_none() {
            let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
            self.notification_connection = Some(ws.connect(NOTIFICATION_WS_URL)?);
        }
        let message = format!(r#"{{"op":"sub","topic":"public.{code}.{topic}","cid":"{code}"}}"#);
        self.notification_connection
            .as_mut()
            .expect("connection just ensured")
            .send(&message)?;
        if !self.sub_messages.contains(&message) {
            self.sub_messages.push(message);
        }
        if !self
            .derivatives_channels
            .iter()
            .any(|(c, k)| c == &code && *k == channel)
        {
            self.derivatives_channels.push((code.clone(), channel));
        }
        if !self.subscriptions.iter().any(|(w, _)| w == &code) {
            self.subscriptions.push((code, symbol.clone()));
        }
        Ok(())
    }

    /// The current open interest
    /// (`GET /linear-swap-api/v1/swap_open_interest`).
    ///
    /// Polled, not pushed. The figure carried is `volume`, the contract count,
    /// which is the unit this client's swap stream and order path both use.
    ///
    /// HTX's field names invert between endpoints, which is worth knowing
    /// before reading either: on this endpoint `volume` is contracts and
    /// `amount` is base currency, while on the trade stream `amount` is the
    /// contract count. Taking `amount` here would silently switch units.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, if the request fails, or if the
    /// venue returns no data point.
    pub fn open_interest(&self, symbol: &Symbol) -> Result<OpenInterest> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "HTX",
                "open interest on a spot client",
                "open interest is a swap figure",
            ));
        }
        let query = format!("contract_code={}", Self::contract_code(symbol));
        let value = self.get("/linear-swap-api/v1/swap_open_interest", &query)?;
        let point = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .and_then(|rows: &Vec<serde_json::Value>| rows.first())
            .ok_or_else(|| Error::NotFound("no open-interest data point".to_string()))?;
        let contracts = point
            .get("volume")
            .and_then(serde_json::Value::as_f64)
            .and_then(|v| parse_decimal(&v.to_string()).ok())
            .ok_or_else(|| Error::Deserialization("no open-interest volume".to_string()))?;
        Ok(OpenInterest {
            symbol: symbol.clone(),
            open_interest: contracts,
            timestamp: value
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })
    }

    /// The current long/short positioning
    /// (`GET /linear-swap-api/v1/swap_elite_account_ratio`), most recent point.
    ///
    /// Polled, not pushed. This is HTX's **elite** ratio: the positioning of
    /// its top traders, not of every account, which is the only positioning
    /// figure it publishes. `buy_ratio` and `sell_ratio` do not sum to one --
    /// `locked_ratio` holds the remainder, for accounts holding both sides --
    /// and they are carried as given rather than rescaled, since rescaling
    /// would invent proportions the venue did not publish.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, if the request fails, or if the
    /// venue returns no data point.
    pub fn long_short_ratio(&self, symbol: &Symbol) -> Result<LongShortRatio> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "HTX",
                "long/short positioning on a spot client",
                "positioning is a swap figure",
            ));
        }
        let query = format!("contract_code={}&period=5min", Self::contract_code(symbol));
        let value = self.get("/linear-swap-api/v1/swap_elite_account_ratio", &query)?;
        let point = value
            .get("data")
            .and_then(|d| d.get("list"))
            .and_then(serde_json::Value::as_array)
            .and_then(|rows: &Vec<serde_json::Value>| rows.last())
            .ok_or_else(|| Error::NotFound("no positioning data point".to_string()))?;
        let ratio = |key: &str| {
            point
                .get(key)
                .and_then(serde_json::Value::as_f64)
                .and_then(|v| parse_decimal(&v.to_string()).ok())
        };
        Ok(LongShortRatio {
            symbol: symbol.clone(),
            long_size: ratio("buy_ratio")
                .ok_or_else(|| Error::Deserialization("no buy_ratio".to_string()))?,
            short_size: ratio("sell_ratio")
                .ok_or_else(|| Error::Deserialization("no sell_ratio".to_string()))?,
            timestamp: point
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })
    }

    /// Subscribe to the public trade stream for `symbol`.
    ///
    /// # Errors
    /// Returns [`Error::NotConnected`] if no WebSocket transport is configured.
    pub fn subscribe_trades(&mut self, symbol: &Symbol) -> Result<()> {
        let topic = format!("market.{}.trade.detail", self.stream_symbol(symbol));
        self.subscribe(symbol, &topic)
    }

    /// Subscribe to the order-book stream for `symbol`.
    ///
    /// # Errors
    /// See [`subscribe_trades`](Self::subscribe_trades).
    pub fn subscribe_book(&mut self, symbol: &Symbol) -> Result<()> {
        let topic = format!("market.{}.depth.step0", self.stream_symbol(symbol));
        self.subscribe(symbol, &topic)
    }

    /// Subscribe to the ticker stream for `symbol`.
    ///
    /// # Errors
    /// See [`subscribe_trades`](Self::subscribe_trades).
    pub fn subscribe_ticker(&mut self, symbol: &Symbol) -> Result<()> {
        // The futures quote channel is `detail`; the spot one is `ticker`.
        let channel = if self.is_futures() {
            "detail"
        } else {
            "ticker"
        };
        let topic = format!("market.{}.{channel}", self.stream_symbol(symbol));
        self.subscribe(symbol, &topic)
    }

    /// The symbol form the stream uses: `BTC-USDT` on futures, `btcusdt` on
    /// spot. The REST paths already make the same distinction through
    /// [`contract_code`](Self::contract_code).
    fn stream_symbol(&self, symbol: &Symbol) -> String {
        if self.is_futures() {
            Self::contract_code(symbol)
        } else {
            Self::wire_symbol(symbol)
        }
    }

    fn subscribe(&mut self, symbol: &Symbol, topic: &str) -> Result<()> {
        self.ensure_market_is_routed()?;
        // The frames come back keyed by whatever form the topic used, so the
        // resolve table has to be keyed the same way: `BTC-USDT` on futures,
        // `btcusdt` on spot. Keying it on the spot form would leave every
        // futures frame unresolvable.
        let wire = self.stream_symbol(symbol);
        if self.connection.is_none() {
            let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
            let connection = ws.connect(if self.is_futures() {
                FUTURES_WS_URL
            } else {
                "wss://api.huobi.pro/ws"
            })?;
            self.connection = Some(connection);
        }
        let message = format!(r#"{{"sub":"{topic}","id":"{wire}"}}"#);
        self.connection
            .as_mut()
            .expect("connection just ensured")
            .send(&message)?;
        if !self.sub_messages.contains(&message) {
            self.sub_messages.push(message.clone());
        }
        if !self.subscriptions.iter().any(|(w, _)| w == &wire) {
            self.subscriptions.push((wire, symbol.clone()));
        }
        Ok(())
    }

    /// Drain all stream events available since the last call. Non-blocking.
    /// (The real adapter gunzips HTX frames before they reach the parser.)
    pub fn poll_events(&mut self) -> Vec<Event> {
        let subscriptions: HashMap<String, Symbol> = self.subscriptions.iter().cloned().collect();
        let resolve = |wire: &str| {
            subscriptions
                .get(wire)
                .cloned()
                .unwrap_or_else(|| Symbol::new(wire, ""))
        };
        let channels = self.derivatives_channels.clone();
        let subscribed = |code: &str, channel: DerivativesChannel| {
            channels.iter().any(|(c, k)| c == code && *k == channel)
        };
        let futures = self.is_futures();
        let mut events = Vec::new();
        if let Some(connection) = self.connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                // HTX closes a stream that does not answer its heartbeat, and
                // the reply has to carry the same value back. Best-effort: if
                // the pong cannot be sent the connection is already gone, and
                // the next poll reconnects.
                if let Some(pong) = pong_for(&frame) {
                    let _ = connection.send(&pong);
                    continue;
                }
                if let Ok(mut parsed) = parse_ws_message(&frame, &resolve) {
                    events.append(&mut parsed);
                }
            }
        }
        // Drain the public notification stream: funding and forced orders. It
        // has its own heartbeat, in the `{"op":"ping"}` shape.
        if let Some(connection) = self.notification_connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Some(pong) = pong_for(&frame) {
                    let _ = connection.send(&pong);
                    continue;
                }
                events.append(&mut parse_notification(&frame, &resolve, &subscribed));
            }
        }
        // Drain the private user-data stream, if open. A futures client's
        // frames are the notification socket's `{"op":"notify"}` shape, not the
        // spot v2 `action`/`ch` one.
        if let Some(connection) = self.private_connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Some(pong) = pong_for(&frame) {
                    let _ = connection.send(&pong);
                    continue;
                }
                if futures {
                    events.append(&mut parse_private_notification(&frame));
                } else if let Ok(mut parsed) = parse_ws_message(&frame, &resolve) {
                    events.append(&mut parsed);
                }
            }
        }
        // A dropped private stream is re-subscribed with a fresh v2 auth handshake
        // (the signature is time-bound, so a stale replay would be rejected).
        if self.user_data_active
            && self
                .private_connection
                .as_ref()
                .is_some_and(|c| !c.is_connected())
        {
            events.push(Event::Disconnected);
            self.private_connection = None;
            if self.subscribe_user_data().is_ok() {
                events.push(Event::Reconnected);
            }
        }
        let url = "wss://api.huobi.pro/ws";
        crate::wsutil::reconnect_if_dropped(
            self.ws.as_deref(),
            url,
            &mut self.connection,
            &self.sub_messages,
            &mut events,
        );
        events
    }

    /// Open the private user-data stream (`wss://api.huobi.pro/ws/v2`).
    /// Authenticates with the v2 `req/auth` handshake (signature = base64(HMAC-SHA256)
    /// over `GET\napi.huobi.pro\n/ws/v2\n<sorted query>`), then subscribes to the
    /// `orders#*` and `accounts.update#2` channels. Afterwards
    /// [`poll_events`](Self::poll_events) also surfaces the account's own
    /// [`Event::OrderUpdate`] and [`Event::BalanceUpdate`].
    ///
    /// A dropped private stream is re-subscribed automatically on the next
    /// [`poll_events`](Self::poll_events); call
    /// [`keepalive_user_data`](Self::keepalive_user_data) periodically to keep it
    /// from being dropped for inactivity.
    ///
    /// # Errors
    /// Returns [`Error::InvalidCredentials`] without credentials,
    /// [`Error::NotConnected`] without a WebSocket transport, or another
    /// [`Error`] if the request fails.
    pub fn subscribe_user_data(&mut self) -> Result<()> {
        self.ensure_market_is_routed()?;
        // Spot and futures are separate accounts on separate hosts: a futures
        // order never appears in the spot account, so watching the wrong one is
        // waiting for a fill that cannot arrive.
        if self.is_futures() {
            return self.subscribe_futures_user_data();
        }
        let creds = self.credentials.as_ref().ok_or(Error::InvalidCredentials(
            "user-data stream requires credentials",
        ))?;
        let timestamp = iso8601_no_millis(self.signed_now_ms());
        // The v2 WS auth signs the same canonical form as REST, over `/ws/v2`.
        let mut params = [
            ("accessKey", creds.api_key.as_str()),
            ("signatureMethod", "HmacSHA256"),
            ("signatureVersion", "2.1"),
            ("timestamp", timestamp.as_str()),
        ];
        params.sort_by(|a, b| a.0.cmp(b.0));
        let encoded = params
            .iter()
            .map(|(key, val)| format!("{}={}", encode(key), encode(val)))
            .collect::<Vec<_>>()
            .join("&");
        let canonical = format!("GET\n{HOST}\n/ws/v2\n{encoded}");
        let signature = hmac_sha256_base64(creds.api_secret.as_bytes(), canonical.as_bytes());
        let auth = format!(
            r#"{{"action":"req","ch":"auth","params":{{"authType":"api","accessKey":"{}","signatureMethod":"HmacSHA256","signatureVersion":"2.1","timestamp":"{timestamp}","signature":"{signature}"}}}}"#,
            creds.api_key
        );
        let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
        let mut connection = ws.connect("wss://api.huobi.pro/ws/v2")?;
        connection.send(&auth)?;
        connection.send(r#"{"action":"sub","ch":"orders#*"}"#)?;
        connection.send(r#"{"action":"sub","ch":"accounts.update#2"}"#)?;
        self.private_connection = Some(connection);
        self.user_data_active = true;
        Ok(())
    }

    /// Open the **futures** private stream on the notification socket.
    ///
    /// The same host the public derivatives channels use
    /// (`api.hbdm.com/linear-swap-notification`), separated by authentication
    /// rather than by URL: an `{"op":"auth"}` frame signed the way a REST call
    /// is, then `orders_cross.*` and `accounts_cross.*` for every contract and
    /// every margin account.
    ///
    /// The socket pings every five seconds and closes when that goes
    /// unanswered; [`poll_events`](Self::poll_events) answers it while draining.
    ///
    /// # Errors
    /// Returns [`Error::InvalidCredentials`] without credentials,
    /// [`Error::NotConnected`] without a WebSocket transport, or another
    /// [`Error`] if the subscription fails.
    fn subscribe_futures_user_data(&mut self) -> Result<()> {
        let creds = self.credentials.as_ref().ok_or(Error::InvalidCredentials(
            "user-data stream requires credentials",
        ))?;
        let timestamp = iso8601_no_millis(self.signed_now_ms());
        // The futures WS auth signs the same canonical form as a REST call,
        // over the socket's own path.
        let mut params = [
            ("AccessKeyId", creds.api_key.as_str()),
            ("SignatureMethod", "HmacSHA256"),
            ("SignatureVersion", "2"),
            ("Timestamp", timestamp.as_str()),
        ];
        params.sort_by(|a, b| a.0.cmp(b.0));
        let encoded = params
            .iter()
            .map(|(key, val)| format!("{}={}", encode(key), encode(val)))
            .collect::<Vec<_>>()
            .join("&");
        let canonical = format!("GET\napi.hbdm.com\n/linear-swap-notification\n{encoded}");
        let signature = hmac_sha256_base64(creds.api_secret.as_bytes(), canonical.as_bytes());
        let auth = format!(
            r#"{{"op":"auth","type":"api","AccessKeyId":"{}","SignatureMethod":"HmacSHA256","SignatureVersion":"2","Timestamp":"{timestamp}","Signature":"{signature}"}}"#,
            creds.api_key
        );
        let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
        let mut connection = ws.connect(NOTIFICATION_WS_URL)?;
        connection.send(&auth)?;
        // `*` is every contract on the cross-margin account, which is the
        // account this client's orders are placed on.
        connection.send(r#"{"op":"sub","topic":"orders_cross.*","cid":"orders"}"#)?;
        connection.send(r#"{"op":"sub","topic":"accounts_cross.*","cid":"accounts"}"#)?;
        self.private_connection = Some(connection);
        self.user_data_active = true;
        Ok(())
    }

    /// Send the HTX v2 application-level heartbeat (`{"action":"ping"}`) on the
    /// private stream so it is not dropped for inactivity. A no-op before
    /// [`subscribe_user_data`](Self::subscribe_user_data).
    ///
    /// # Errors
    /// Returns an [`Error`] if the ping cannot be sent.
    pub fn keepalive_user_data(&mut self) -> Result<()> {
        if let Some(connection) = self.private_connection.as_mut() {
            connection.send(r#"{"action":"ping"}"#)?;
        }
        Ok(())
    }

    /// The spot `account-id`, fetched once and cached.
    fn account_id(&self) -> Result<String> {
        if let Some(id) = self.account_id.borrow().clone() {
            return Ok(id);
        }
        let value = self.signed_request(HttpMethod::Get, "/v1/account/accounts", &[], "")?;
        let accounts = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing accounts".to_string()))?;
        let id = accounts
            .iter()
            .find(|a| a.get("type").and_then(serde_json::Value::as_str) == Some("spot"))
            .and_then(|a| a.get("id"))
            .map(|id| {
                id.as_u64()
                    .map_or_else(|| id.as_str().unwrap_or("").to_string(), |n| n.to_string())
            })
            .ok_or_else(|| Error::NotFound("no spot account".to_string()))?;
        *self.account_id.borrow_mut() = Some(id.clone());
        Ok(id)
    }

    /// Place an order.
    ///
    /// # Errors
    /// Returns an [`Error`] if the order is invalid, credentials are missing, or
    /// the venue rejects it.
    /// Refuse `reduce_only` where there is no position to reduce.
    ///
    /// A spot account holds balances, not positions, so a reduce-only order has
    /// nothing to close. Dropping the flag would place an opening order where a
    /// closing one was asked for -- the opposite trade -- so it is refused, the
    /// way Binance's spot path and Upbit already refuse it.
    fn ensure_reduce_only_is_reducible(&self, request: &OrderRequest) -> Result<()> {
        if request.reduce_only && !self.is_futures() {
            return Err(Error::unsupported_field(
                "HTX",
                "reduce_only on a spot order",
                "spot holds balances, not positions, and has none to reduce",
            ));
        }
        Ok(())
    }

    /// Place a trigger order through HTX's algo endpoints.
    ///
    /// Spot uses `/v2/algo-orders` and the swap `swap_trigger_order`; neither
    /// takes a trigger on the ordinary order call.
    ///
    /// `trigger_type` is the comparison and HTX will not infer it: `le` fires at
    /// or below the trigger, which protects a long, and `ge` at or above, which
    /// covers a short. The wrong one arms the order on the side that never
    /// comes.
    ///
    /// The swap path spells "take the market" as an order price type
    /// (`optimal_5`) rather than as a price, so a stop-market sends that and a
    /// stop-limit sends `limit` with its price. Spot's algo endpoint takes an
    /// `orderType` the same way.
    ///
    /// # Errors
    /// Returns an [`Error`] if the trigger price is missing or the request
    /// fails.
    fn place_trigger_order(&self, request: &OrderRequest) -> Result<Order> {
        let stop = request
            .stop_price
            .ok_or(Error::InvalidOrder("a trigger order requires a stop price"))?;
        let (path, mut body) = if self.is_futures() {
            let volume = decimal_to_contracts(request.quantity)?;
            (
                "/linear-swap-api/v1/swap_trigger_order",
                serde_json::json!({
                    "contract_code": Self::contract_code(&request.symbol),
                    // Selling to protect a long fires on the way down; buying to
                    // cover a short fires on the way up. The swap asks for the
                    // comparison outright and will not infer it.
                    "trigger_type": match request.side {
                        OrderSide::Sell => "le",
                        OrderSide::Buy => "ge",
                    },
                    "trigger_price": format_decimal(stop),
                    "volume": volume,
                    "direction": side_str(request.side),
                    "offset": if request.reduce_only { "close" } else { "open" },
                    "lever_rate": self.leverage.get(),
                    // "Take the market" is an order *price type* here, not a
                    // price: `optimal_5` sweeps the best five levels.
                    "order_price_type": if request.price.is_some() {
                        "limit"
                    } else {
                        "optimal_5"
                    },
                }),
            )
        } else {
            (
                "/v2/algo-orders",
                serde_json::json!({
                    "accountId": self.account_id()?,
                    "symbol": Self::wire_symbol(&request.symbol),
                    // Spot reads the direction off the side rather than taking a
                    // comparison of its own.
                    "orderSide": side_str(request.side),
                    "orderType": order_type_str(request.order_type),
                    "orderSize": format_decimal(request.quantity),
                    "stopPrice": format_decimal(stop),
                    // Mandatory: a spot algo order is cancelled by this id, not
                    // by an exchange-assigned one, so an unnamed order would be
                    // unreachable afterwards.
                    "clientOrderId": request
                        .client_order_id
                        .clone()
                        .unwrap_or_else(|| self.client_ids.next_id()),
                }),
            )
        };
        // Only a limit trigger has a price. An empty string in the price slot is
        // not "no price" to either endpoint -- it is a rejected order.
        if let Some(price) = request.price {
            let key = if self.is_futures() {
                "order_price"
            } else {
                "orderPrice"
            };
            body[key] = serde_json::json!(format_decimal(price));
        }
        let value = self.signed_request(HttpMethod::Post, path, &[], &body.to_string())?;
        let data = value
            .get("data")
            .ok_or_else(|| Error::Deserialization("missing order data".to_string()))?;
        // The swap answers with an exchange order id; the spot algo endpoint
        // echoes back the client order id, which is the only handle it has.
        let id = data
            .get("order_id_str")
            .or_else(|| data.get("order_id"))
            .or_else(|| data.get("clientOrderId"))
            .map_or_else(
                || data.as_str().unwrap_or_default().to_string(),
                |id| id.as_str().map_or_else(|| id.to_string(), str::to_string),
            );
        Ok(Order {
            id,
            client_order_id: request.client_order_id.clone(),
            symbol: request.symbol.clone(),
            side: request.side,
            order_type: request.order_type,
            status: OrderStatus::New,
            quantity: request.quantity,
            filled_quantity: Decimal::ZERO,
            price: request.price,
            average_price: None,
        })
    }

    pub fn place_order(&self, request: &OrderRequest) -> Result<Order> {
        self.ensure_reduce_only_is_reducible(request)?;
        request.validate()?;
        // A trigger is a different endpoint on both markets.
        if request.order_type.is_trigger() {
            return self.place_trigger_order(request);
        }
        if self.is_futures() {
            return self.place_futures_order(request);
        }
        // Resolved before the account lookup, so a request this venue cannot
        // express costs no round-trip and leaves nothing on the wire.
        ensure_no_stp(request)?;
        let order_kind = format!("{}-{}", side_str(request.side), order_kind_str(request)?);
        let account_id = self.account_id()?;
        let mut body = serde_json::json!({
            "account-id": account_id,
            "symbol": Self::wire_symbol(&request.symbol),
            "type": order_kind,
            "amount": format_decimal(request.quantity),
        });
        if let Some(price) = request.price {
            body["price"] = serde_json::json!(format_decimal(price));
        }
        if let Some(id) = &request.client_order_id {
            body["client-order-id"] = serde_json::json!(id.clone());
        }
        let value = self.signed_request(
            HttpMethod::Post,
            "/v1/order/orders/place",
            &[],
            &body.to_string(),
        )?;
        let order_id = value
            .get("data")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| Error::Deserialization("missing order id".to_string()))?;
        Ok(Order {
            id: order_id.to_string(),
            client_order_id: request.client_order_id.clone(),
            symbol: request.symbol.clone(),
            side: request.side,
            order_type: request.order_type,
            status: OrderStatus::New,
            quantity: request.quantity,
            filled_quantity: Decimal::ZERO,
            price: request.price,
            average_price: None,
        })
    }

    /// Cancel an open order by venue id.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the venue rejects it.
    pub fn cancel_order(&self, symbol: &Symbol, order_id: &str) -> Result<()> {
        if self.is_futures() {
            let body = serde_json::json!({
                "contract_code": Self::contract_code(symbol),
                "order_id": order_id,
            });
            self.signed_request(
                HttpMethod::Post,
                "/linear-swap-api/v1/swap_cross_cancel",
                &[],
                &body.to_string(),
            )?;
            return Ok(());
        }
        let path = format!("/v1/order/orders/{order_id}/submitcancel");
        self.signed_request(HttpMethod::Post, &path, &[], "{}")?;
        Ok(())
    }

    /// Query a single order by venue id.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the order is unknown.
    pub fn query_order(&self, symbol: &Symbol, order_id: &str) -> Result<Order> {
        if self.is_futures() {
            let body = serde_json::json!({
                "contract_code": Self::contract_code(symbol),
                "order_id": order_id,
            });
            let value = self.signed_request(
                HttpMethod::Post,
                "/linear-swap-api/v1/swap_cross_order_info",
                &[],
                &body.to_string(),
            )?;
            let order = value
                .get("data")
                .and_then(serde_json::Value::as_array)
                .and_then(|a| a.first())
                .ok_or_else(|| Error::NotFound(format!("order {order_id}")))?;
            return swap_order_from_value(symbol.clone(), order);
        }
        let path = format!("/v1/order/orders/{order_id}");
        let value = self.signed_request(HttpMethod::Get, &path, &[], "")?;
        let data = value
            .get("data")
            .ok_or_else(|| Error::Deserialization("missing order data".to_string()))?;
        order_from_value(symbol.clone(), data)
    }

    /// Open orders, optionally filtered to one `symbol`.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn open_orders(&self, symbol: Option<&Symbol>) -> Result<Vec<Order>> {
        if self.is_futures() {
            let sym = symbol.ok_or(Error::InvalidOrder(
                "HTX futures open_orders requires a symbol",
            ))?;
            let body = serde_json::json!({ "contract_code": Self::contract_code(sym) });
            let value = self.signed_request(
                HttpMethod::Post,
                "/linear-swap-api/v1/swap_cross_openorders",
                &[],
                &body.to_string(),
            )?;
            let orders = value
                .get("data")
                .and_then(|d| d.get("orders"))
                .and_then(serde_json::Value::as_array)
                .ok_or_else(|| Error::Deserialization("missing orders".to_string()))?;
            return orders
                .iter()
                .map(|order| swap_order_from_value(sym.clone(), order))
                .collect();
        }
        let account_id = self.account_id()?;
        let wire = symbol.map(Self::wire_symbol);
        let mut params: Vec<(&str, &str)> = vec![("account-id", &account_id)];
        if let Some(w) = &wire {
            params.push(("symbol", w));
        }
        let value = self.signed_request(HttpMethod::Get, "/v1/order/openOrders", &params, "")?;
        let data = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing orders".to_string()))?;
        data.iter()
            .map(|order| {
                let sym = symbol.cloned().unwrap_or_else(|| {
                    let raw = order
                        .get("symbol")
                        .and_then(serde_json::Value::as_str)
                        .unwrap_or("");
                    split_wire_symbol(raw)
                });
                order_from_value(sym, order)
            })
            .collect()
    }

    /// Spot account balances (HTX reports `trade` and `frozen` entries per asset).
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn balances(&self) -> Result<Vec<Balance>> {
        if self.is_futures() {
            return self.futures_balances();
        }
        let account_id = self.account_id()?;
        let path = format!("/v1/account/accounts/{account_id}/balance");
        let value = self.signed_request(HttpMethod::Get, &path, &[], "")?;
        let list = value
            .get("data")
            .and_then(|d| d.get("list"))
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing balance list".to_string()))?;
        let mut merged: HashMap<String, (Decimal, Decimal)> = HashMap::new();
        for entry in list {
            let currency = entry
                .get("currency")
                .and_then(serde_json::Value::as_str)
                .unwrap_or("");
            let kind = entry
                .get("type")
                .and_then(serde_json::Value::as_str)
                .unwrap_or("");
            let amount = decimal_field(entry, "balance").unwrap_or(Decimal::ZERO);
            let slot = merged.entry(currency.to_string()).or_default();
            if kind == "frozen" {
                slot.1 += amount;
            } else {
                slot.0 += amount;
            }
        }
        let mut balances: Vec<Balance> = merged
            .into_iter()
            .map(|(asset, (free, locked))| Balance {
                asset,
                free,
                locked,
            })
            .collect();
        balances.sort_by(|a, b| a.asset.cmp(&b.asset));
        Ok(balances)
    }

    /// Place a USDT-margined swap order via the cross-margin family
    /// (`/linear-swap-api/v1/swap_cross_order`). Orders carry an integer contract
    /// `volume`, a `direction` (buy/sell), an `offset` (open/close; reduce-only
    /// closes), the stored `lever_rate`, and an `order_price_type`
    /// (`optimal_5` for market, `limit` otherwise).
    fn place_futures_order(&self, request: &OrderRequest) -> Result<Order> {
        let volume = decimal_to_contracts(request.quantity)?;
        let mut body = serde_json::json!({
            "contract_code": Self::contract_code(&request.symbol),
            "direction": side_str(request.side),
            "offset": if request.reduce_only { "close" } else { "open" },
            "lever_rate": self.leverage.get(),
            "volume": volume,
        });
        match request.order_type {
            OrderType::Market | OrderType::StopMarket => {
                // `optimal_5` sweeps the best five levels and cancels the rest,
                // which is the venue's market order and already IOC-shaped;
                // there is no fill-or-kill spelling for it.
                if request.time_in_force == TimeInForce::Fok {
                    return Err(Error::unsupported_field(
                        "HTX",
                        "a FOK time-in-force on a futures market order",
                        "the futures market order is `optimal_5`, which is IOC-shaped",
                    ));
                }
                body["order_price_type"] = serde_json::json!("optimal_5");
            }
            OrderType::Limit | OrderType::StopLimit => {
                let price = request
                    .price
                    .ok_or(Error::InvalidOrder("limit order requires a price"))?;
                body["order_price_type"] = serde_json::json!(futures_price_type(request)?);
                body["price"] = serde_json::json!(format_decimal(price));
            }
        }
        if let Some(id) = &request.client_order_id {
            if let Ok(numeric) = id.parse::<u64>() {
                body["client_order_id"] = serde_json::json!(numeric);
            }
        }
        let value = self.signed_request(
            HttpMethod::Post,
            "/linear-swap-api/v1/swap_cross_order",
            &[],
            &body.to_string(),
        )?;
        let data = value
            .get("data")
            .ok_or_else(|| Error::Deserialization("missing order data".to_string()))?;
        let order_id = data
            .get("order_id_str")
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
            .or_else(|| {
                data.get("order_id")
                    .and_then(serde_json::Value::as_u64)
                    .map(|n| n.to_string())
            })
            .ok_or_else(|| Error::Deserialization("missing order id".to_string()))?;
        Ok(Order {
            id: order_id,
            client_order_id: request.client_order_id.clone(),
            symbol: request.symbol.clone(),
            side: request.side,
            order_type: request.order_type,
            status: OrderStatus::New,
            quantity: request.quantity,
            filled_quantity: Decimal::ZERO,
            price: request.price,
            average_price: None,
        })
    }

    /// USDT-margined cross account balance
    /// (`/linear-swap-api/v1/swap_cross_account_info`).
    fn futures_balances(&self) -> Result<Vec<Balance>> {
        let value = self.signed_request(
            HttpMethod::Post,
            "/linear-swap-api/v1/swap_cross_account_info",
            &[],
            "{}",
        )?;
        let list = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing account data".to_string()))?;
        Ok(list
            .iter()
            .map(|acct| {
                let asset = acct
                    .get("margin_asset")
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or("USDT")
                    .to_string();
                Balance {
                    asset,
                    free: decimal_field(acct, "margin_available").unwrap_or(Decimal::ZERO),
                    locked: decimal_field(acct, "margin_frozen").unwrap_or(Decimal::ZERO),
                }
            })
            .collect())
    }

    /// Open positions on the USDT-margined cross account
    /// (`/linear-swap-api/v1/swap_cross_position_info`).
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn positions(&self, symbol: Option<&Symbol>) -> Result<Vec<Position>> {
        let body = symbol.map_or_else(
            || "{}".to_string(),
            |s| format!(r#"{{"contract_code":"{}"}}"#, Self::contract_code(s)),
        );
        let value = self.signed_request(
            HttpMethod::Post,
            "/linear-swap-api/v1/swap_cross_position_info",
            &[],
            &body,
        )?;
        let list = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing position data".to_string()))?;
        list.iter().map(parse_futures_position).collect()
    }

    /// Set the leverage for `symbol`
    /// (`/linear-swap-api/v1/swap_cross_switch_lever_rate`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the leverage is rejected or the request fails.
    pub fn set_leverage(&self, symbol: &Symbol, leverage: u32) -> Result<()> {
        let lever = leverage.max(1);
        self.leverage.set(lever);
        let body = format!(
            r#"{{"contract_code":"{}","lever_rate":{lever}}}"#,
            Self::contract_code(symbol)
        );
        self.signed_request(
            HttpMethod::Post,
            "/linear-swap-api/v1/swap_cross_switch_lever_rate",
            &[],
            &body,
        )?;
        Ok(())
    }

    /// Set the margin mode for `symbol`.
    ///
    /// This binding drives HTX's **cross-margin** swap family, so `Cross` is a
    /// no-op success. HTX does not expose a per-symbol switch to isolated within
    /// this family (it is an account/order-family choice), so `Isolated` returns
    /// an [`Error::Exchange`] documenting the limitation.
    ///
    /// # Errors
    /// Returns [`Error::Exchange`] when `Isolated` is requested.
    pub fn set_margin_mode(&self, _symbol: &Symbol, mode: MarginMode) -> Result<()> {
        match mode {
            MarginMode::Cross => Ok(()),
            MarginMode::Isolated => Err(Error::Exchange {
                code: "unsupported".to_string(),
                message: "HTX linear-swap binding uses the cross-margin family; \
                          isolated is not switchable per-symbol"
                    .to_string(),
            }),
        }
    }

    /// Flatten the open position in `symbol` with a reduce-only market order.
    ///
    /// # Errors
    /// Returns [`Error::NotFound`] if there is no open position, or another
    /// [`Error`] if the request fails.
    pub fn close_position(&self, symbol: &Symbol) -> Result<Order> {
        let position = self
            .positions(Some(symbol))?
            .into_iter()
            .find(|p| &p.symbol == symbol)
            .ok_or_else(|| Error::NotFound(format!("no open position for {symbol}")))?;
        let request = match position.side {
            PositionSide::Long => OrderRequest::market_sell(symbol.clone(), position.quantity),
            PositionSide::Short => OrderRequest::market_buy(symbol.clone(), position.quantity),
        }
        .reduce_only();
        self.place_order(&request)
    }

    fn get(&self, path: &str, query: &str) -> Result<serde_json::Value> {
        self.ensure_market_is_routed()?;
        let url = format!("{}{path}?{query}", self.rest_base);
        let response = self.http.execute(&HttpRequest::get(url))?;
        unwrap_status(&response.body).map_err(|e| {
            super::attribute_to_http_status(&response, e).with_retry_after(response.retry_after())
        })
    }

    /// Sign AWS-style and issue the request. Auth params live in the query and are
    /// the only signed material (a POST body is sent unsigned as JSON).
    fn signed_request(
        &self,
        method: HttpMethod,
        path: &str,
        extra: &[(&str, &str)],
        body: &str,
    ) -> Result<serde_json::Value> {
        self.ensure_market_is_routed()?;
        let creds = self.credentials.as_ref().ok_or(Error::InvalidCredentials(
            "signed endpoint requires credentials",
        ))?;
        let mut params: Vec<(String, String)> = vec![
            ("AccessKeyId".to_string(), creds.api_key.clone()),
            ("SignatureMethod".to_string(), "HmacSHA256".to_string()),
            ("SignatureVersion".to_string(), "2".to_string()),
            (
                "Timestamp".to_string(),
                iso8601_no_millis(self.signed_now_ms()),
            ),
        ];
        for (key, val) in extra {
            params.push(((*key).to_string(), (*val).to_string()));
        }
        params.sort_by(|a, b| a.0.cmp(&b.0));
        let encoded = params
            .iter()
            .map(|(key, val)| format!("{}={}", encode(key), encode(val)))
            .collect::<Vec<_>>()
            .join("&");
        let canonical = format!("{}\n{}\n{path}\n{encoded}", method.as_str(), self.host);
        let signature = hmac_sha256_base64(creds.api_secret.as_bytes(), canonical.as_bytes());
        let url = format!(
            "{}{path}?{encoded}&Signature={}",
            self.rest_base,
            encode(&signature)
        );
        let mut request = HttpRequest::new(method, url);
        if !body.is_empty() {
            request = request
                .with_header("Content-Type", "application/json")
                .with_body(body.to_string());
        }
        let response = self.http.execute(&request)?;
        unwrap_status(&response.body).map_err(|e| {
            super::attribute_to_http_status(&response, e).with_retry_after(response.retry_after())
        })
    }
}

fn map_period(interval: &str) -> String {
    match interval {
        "1m" => "1min",
        "5m" => "5min",
        "15m" => "15min",
        "30m" => "30min",
        "1h" => "60min",
        "4h" => "4hour",
        "1d" => "1day",
        "1w" => "1week",
        other => other,
    }
    .to_string()
}

/// Unwrap either of HTX's two response envelopes.
///
/// The v1 and swap endpoints answer `{"status":"ok",...}`; the v2 endpoints --
/// among them the spot algo orders a trigger is placed through -- answer
/// `{"code":200,"message":"...",...}` and carry no `status` at all. Reading only
/// the first turns every v2 success into an error with an empty code and an
/// empty message, which is how a placed trigger order looks like a failed one.
fn unwrap_status(body: &str) -> Result<serde_json::Value> {
    let value: serde_json::Value =
        serde_json::from_str(body).map_err(|e| Error::Deserialization(e.to_string()))?;
    if value.get("status").and_then(serde_json::Value::as_str) == Some("ok") {
        Ok(value)
    } else if let Some(code) = value.get("code").and_then(serde_json::Value::as_u64) {
        if code == 200 {
            return Ok(value);
        }
        let message = value
            .get("message")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("");
        Err(map_error(&code.to_string(), message))
    } else {
        let code = value
            .get("err-code")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("");
        let message = value
            .get("err-msg")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("");
        Err(map_error(code, message))
    }
}

fn map_error(code: &str, message: &str) -> Error {
    match code {
        "request-rate-limit" | "rate-limit" => Error::RateLimited { retry_after: None },
        "api-signature-not-valid" | "api-key-not-valid" | "invalid-signature" => {
            Error::Auth(message.to_string())
        }
        "account-frozen-balance-insufficient-error"
        | "order-value-min-error"
        | "insufficient-balance" => Error::InsufficientBalance,
        "base-symbol-error" | "invalid-symbol" => Error::InvalidSymbol(message.to_string()),
        "base-record-invalid" | "order-not-found" => Error::NotFound(message.to_string()),
        other => Error::Exchange {
            code: other.to_string(),
            message: message.to_string(),
        },
    }
}

fn side_str(side: OrderSide) -> &'static str {
    match side {
        OrderSide::Buy => "buy",
        OrderSide::Sell => "sell",
    }
}

/// HTX spells the spot order kind, its time-in-force and post-only as one
/// `<side>-<kind>` string (`buy-limit`, `buy-ioc`, `buy-limit-maker`,
/// `buy-limit-fok`), so those three settings share a single slot and a
/// combination needing two of them is refused rather than resolved by dropping
/// one. There is no market fill-or-kill, so that is refused too.
fn order_kind_str(request: &OrderRequest) -> Result<&'static str> {
    match (request.order_type, request.time_in_force, request.post_only) {
        (OrderType::Market | OrderType::StopMarket, TimeInForce::Fok, _) => {
            Err(Error::unsupported_field(
                "HTX",
                "a FOK time-in-force on a market order",
                "HTX has no `<side>-market-fok` order kind",
            ))
        }
        (OrderType::Market | OrderType::StopMarket, _, true) => Err(Error::unsupported_field(
            "HTX",
            "post_only on a market order",
            "HTX's post-only is the `limit-maker` order kind",
        )),
        (OrderType::Market | OrderType::StopMarket, _, false)
        | (OrderType::Limit | OrderType::StopLimit, TimeInForce::Gtc, false) => {
            Ok(order_type_str(request.order_type))
        }
        (OrderType::Limit | OrderType::StopLimit, TimeInForce::Gtc, true) => Ok("limit-maker"),
        (OrderType::Limit | OrderType::StopLimit, _, true) => Err(Error::unsupported_field(
            "HTX",
            "post_only together with a non-GTC time-in-force",
            "both live in the one `<side>-<kind>` slot",
        )),
        (OrderType::Limit | OrderType::StopLimit, TimeInForce::Ioc, false) => Ok("ioc"),
        (OrderType::Limit | OrderType::StopLimit, TimeInForce::Fok, false) => Ok("limit-fok"),
    }
}

/// The HTX swap `order_price_type`. Post-only and IOC/FOK are values of that one
/// field, so they cannot both be spelled and a conflict is refused.
fn futures_price_type(request: &OrderRequest) -> Result<&'static str> {
    match (request.time_in_force, request.post_only) {
        (TimeInForce::Gtc, false) => Ok("limit"),
        (TimeInForce::Gtc, true) => Ok("post_only"),
        (TimeInForce::Ioc, false) => Ok("ioc"),
        (TimeInForce::Fok, false) => Ok("fok"),
        (_, true) => Err(Error::unsupported_field(
            "HTX",
            "post_only together with a non-GTC time-in-force",
            "both are values of the one `order_price_type` field",
        )),
    }
}

/// HTX's spot order API carries no self-trade-prevention field, so a policy the
/// caller set cannot be honoured and the order is refused instead of placed
/// without it.
fn ensure_no_stp(request: &OrderRequest) -> Result<()> {
    if request.stp == SelfTradePrevention::None {
        return Ok(());
    }
    Err(Error::unsupported_field(
        "HTX",
        "a self-trade-prevention policy",
        "HTX's spot order API carries no STP field",
    ))
}

fn order_type_str(order_type: OrderType) -> &'static str {
    match order_type {
        OrderType::Market | OrderType::StopMarket => "market",
        OrderType::Limit | OrderType::StopLimit => "limit",
    }
}

fn parse_side(raw: &str) -> Result<OrderSide> {
    // HTX order `type` is `<side>-<kind>` (e.g. `buy-limit`).
    match raw.split('-').next() {
        Some("buy") => Ok(OrderSide::Buy),
        Some("sell") => Ok(OrderSide::Sell),
        _ => Err(Error::Deserialization(format!(
            "unknown order type {raw:?}"
        ))),
    }
}

fn parse_order_type(raw: &str) -> Result<OrderType> {
    if raw.contains("market") {
        Ok(OrderType::Market)
    } else if raw.contains("limit") {
        Ok(OrderType::Limit)
    } else {
        Err(Error::Deserialization(format!(
            "unknown order type {raw:?}"
        )))
    }
}

fn parse_state(raw: &str) -> Result<OrderStatus> {
    match raw {
        "submitted" | "created" => Ok(OrderStatus::New),
        "partial-filled" => Ok(OrderStatus::PartiallyFilled),
        "filled" => Ok(OrderStatus::Filled),
        "canceled" | "partial-canceled" | "cancelling" => Ok(OrderStatus::Canceled),
        other => Err(Error::Deserialization(format!("unknown state {other:?}"))),
    }
}

fn nonzero_decimal(raw: &str) -> Option<Decimal> {
    crate::normalize::parse_opt_decimal(Some(raw))
        .ok()
        .flatten()
        .filter(|d| *d > Decimal::ZERO)
}

/// Read a decimal that HTX may encode as a JSON number or string.
fn decimal_field(value: &serde_json::Value, key: &str) -> Result<Decimal> {
    let field = value
        .get(key)
        .ok_or_else(|| Error::Deserialization(format!("missing field {key:?}")))?;
    decimal_value(field)
}

fn decimal_value(field: &serde_json::Value) -> Result<Decimal> {
    match field {
        serde_json::Value::String(s) => parse_decimal(s),
        serde_json::Value::Number(n) => parse_decimal(&n.to_string()),
        other => Err(Error::Deserialization(format!("not a number: {other}"))),
    }
}

/// Read `value[key][index]` as a decimal (HTX `bid`/`ask` are `[price, size]`).
fn decimal_at(value: &serde_json::Value, key: &str, index: usize) -> Result<Decimal> {
    let field = value
        .get(key)
        .and_then(serde_json::Value::as_array)
        .and_then(|a| a.get(index))
        .ok_or_else(|| Error::Deserialization(format!("missing {key}[{index}]")))?;
    decimal_value(field)
}

fn f64_field(value: &serde_json::Value, key: &str) -> Result<f64> {
    let field = value
        .get(key)
        .ok_or_else(|| Error::Deserialization(format!("missing field {key:?}")))?;
    field
        .as_f64()
        .or_else(|| field.as_str().and_then(|s| s.parse().ok()))
        .ok_or_else(|| Error::Deserialization(format!("field {key:?} not a number")))
}

fn num_levels(value: Option<&serde_json::Value>) -> Result<Vec<BookLevel>> {
    let array = value
        .and_then(serde_json::Value::as_array)
        .ok_or_else(|| Error::Deserialization("missing depth levels".to_string()))?;
    array
        .iter()
        .map(|level| {
            let pair = level
                .as_array()
                .ok_or_else(|| Error::Deserialization("depth level not an array".to_string()))?;
            let price = decimal_value(
                pair.first()
                    .ok_or_else(|| Error::Deserialization("depth price missing".to_string()))?,
            )?;
            let quantity =
                decimal_value(pair.get(1).ok_or_else(|| {
                    Error::Deserialization("depth quantity missing".to_string())
                })?)?;
            Ok(BookLevel { price, quantity })
        })
        .collect()
}

fn parse_kline_obj(obj: &serde_json::Value) -> Result<Candle> {
    let ts = obj
        .get("id")
        .and_then(serde_json::Value::as_i64)
        .ok_or_else(|| Error::Deserialization("kline id missing".to_string()))?;
    Candle::new(
        f64_field(obj, "open")?,
        f64_field(obj, "high")?,
        f64_field(obj, "low")?,
        f64_field(obj, "close")?,
        f64_field(obj, "amount")?,
        ts,
    )
    .map_err(|e| Error::Deserialization(e.to_string()))
}

fn order_from_value(symbol: Symbol, data: &serde_json::Value) -> Result<Order> {
    let kind = data
        .get("type")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let state = data
        .get("state")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let id = data
        .get("id")
        .map(|id| {
            id.as_u64()
                .map_or_else(|| id.as_str().unwrap_or("").to_string(), |n| n.to_string())
        })
        .unwrap_or_default();
    let client_order_id = data
        .get("client-order-id")
        .and_then(serde_json::Value::as_str)
        .filter(|s| !s.is_empty())
        .map(str::to_string);
    let filled = decimal_field(data, "field-amount").unwrap_or(Decimal::ZERO);
    Ok(Order {
        id,
        client_order_id,
        symbol,
        side: parse_side(kind)?,
        order_type: parse_order_type(kind)?,
        status: parse_state(state)?,
        quantity: decimal_field(data, "amount")?,
        filled_quantity: filled,
        price: nonzero_decimal(
            &decimal_field(data, "price")
                .unwrap_or(Decimal::ZERO)
                .to_string(),
        ),
        average_price: None,
    })
}

/// Map an HTX linear-swap order `status` integer to the unified status.
fn swap_status(code: i64) -> OrderStatus {
    match code {
        3 => OrderStatus::PartiallyFilled,
        6 => OrderStatus::Filled,
        4 | 7 => OrderStatus::Canceled,
        _ => OrderStatus::New,
    }
}

/// Parse an HTX linear-swap order object (`swap_cross_order_info` / openorders).
fn swap_order_from_value(symbol: Symbol, data: &serde_json::Value) -> Result<Order> {
    let id = data
        .get("order_id_str")
        .and_then(serde_json::Value::as_str)
        .map(str::to_string)
        .or_else(|| {
            data.get("order_id")
                .and_then(serde_json::Value::as_i64)
                .map(|n| n.to_string())
        })
        .unwrap_or_default();
    let direction = data
        .get("direction")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let price_type = data
        .get("order_price_type")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let order_type = if price_type.contains("limit") || price_type == "post_only" {
        OrderType::Limit
    } else {
        OrderType::Market
    };
    let status = swap_status(
        data.get("status")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0),
    );
    let filled = decimal_field(data, "trade_volume").unwrap_or(Decimal::ZERO);
    let avg = decimal_field(data, "trade_avg_price")
        .ok()
        .filter(|d| *d > Decimal::ZERO);
    let price = decimal_field(data, "price")
        .ok()
        .filter(|d| *d > Decimal::ZERO);
    Ok(Order {
        id,
        client_order_id: None,
        symbol,
        side: parse_side(direction)?,
        order_type,
        status,
        quantity: decimal_field(data, "volume").unwrap_or(Decimal::ZERO),
        filled_quantity: filled,
        price,
        average_price: avg,
    })
}

const KNOWN_QUOTES: &[&str] = &["usdt", "usdc", "eur", "btc", "eth", "usd"];

fn split_wire_symbol(wire: &str) -> Symbol {
    for quote in KNOWN_QUOTES {
        if let Some(base) = wire.strip_suffix(quote) {
            if !base.is_empty() {
                return Symbol::new(base, *quote);
            }
        }
    }
    Symbol::new(wire, "")
}

/// Build an [`Order`] from an HTX v2 `orders#*` push payload.
fn ws_order_from_data(data: &serde_json::Value) -> Result<Order> {
    let wire = data
        .get("symbol")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::Deserialization("missing order symbol".to_string()))?;
    let order_type = data
        .get("type")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::Deserialization("missing order type".to_string()))?;
    let state = data
        .get("orderStatus")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::Deserialization("missing orderStatus".to_string()))?;
    // HTX encodes the order id as a JSON number.
    let id = match data.get("orderId") {
        Some(serde_json::Value::Number(n)) => n.to_string(),
        Some(serde_json::Value::String(s)) => s.clone(),
        _ => return Err(Error::Deserialization("missing orderId".to_string())),
    };
    let client_id = data
        .get("clientOrderId")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let opt_dec = |key: &str| {
        data.get(key)
            .and_then(|f| decimal_value(f).ok())
            .filter(|d| *d > Decimal::ZERO)
    };
    Ok(Order {
        id,
        client_order_id: (!client_id.is_empty()).then(|| client_id.to_string()),
        symbol: split_wire_symbol(wire),
        side: parse_side(order_type)?,
        order_type: parse_order_type(order_type)?,
        status: parse_state(state)?,
        quantity: decimal_field(data, "orderSize").unwrap_or(Decimal::ZERO),
        filled_quantity: data
            .get("execAmt")
            .and_then(|f| decimal_value(f).ok())
            .unwrap_or(Decimal::ZERO),
        price: opt_dec("orderPrice"),
        average_price: opt_dec("tradePrice"),
    })
}

/// Build a [`Balance`] from an HTX v2 `accounts.update` push payload. `available`
/// is the free balance; `balance` (when present) is the total, so the locked
/// amount is `balance - available`.
fn ws_balance_from_data(data: &serde_json::Value) -> Result<Balance> {
    let currency = data
        .get("currency")
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::Deserialization("missing balance currency".to_string()))?;
    let available = data.get("available").and_then(|f| decimal_value(f).ok());
    let total = data.get("balance").and_then(|f| decimal_value(f).ok());
    let (free, locked) = match (available, total) {
        (Some(free), Some(total)) => (free, (total - free).max(Decimal::ZERO)),
        (Some(free), None) => (free, Decimal::ZERO),
        (None, Some(total)) => (total, Decimal::ZERO),
        (None, None) => (Decimal::ZERO, Decimal::ZERO),
    };
    Ok(Balance {
        asset: currency.to_string(),
        free,
        locked,
    })
}

/// HTX's USDⓈ-M linear-swap WebSocket.
const FUTURES_WS_URL: &str = "wss://api.hbdm.com/linear-swap-ws";

/// The price out of a `[price, size]` pair, or zero.
///
/// The futures quote channel publishes its top of book as two-element arrays
/// where the spot one publishes scalars.
fn level_price(tick: &serde_json::Value, key: &str) -> Decimal {
    tick.get(key)
        .and_then(serde_json::Value::as_array)
        .and_then(|pair| pair.first())
        .and_then(serde_json::Value::as_f64)
        .and_then(|price| parse_decimal(&price.to_string()).ok())
        .unwrap_or(Decimal::ZERO)
}

/// Turn one notification frame into derivatives events.
///
/// The topic is `public.<contract>.<channel>`, and a data frame carries
/// `"op":"notify"`. A subscribe acknowledgement carries `"op":"sub"` and no
/// data, so it falls through to an empty result rather than being mistaken for
/// an empty channel.
/// Read HTX's futures private frames off the notification socket.
///
/// `orders_cross.<contract>` carries one order per frame and
/// `accounts_cross.<account>` one margin account. Both arrive under the same
/// `{"op":"notify"}` envelope as the public channels, so the topic's first
/// segment is what separates them.
fn parse_private_notification(frame: &str) -> Vec<Event> {
    let Ok(value) = serde_json::from_str::<serde_json::Value>(frame) else {
        return Vec::new();
    };
    if value.get("op").and_then(serde_json::Value::as_str) != Some("notify") {
        return Vec::new();
    }
    let Some(topic) = value.get("topic").and_then(serde_json::Value::as_str) else {
        return Vec::new();
    };
    let channel = topic.split('.').next().unwrap_or("");
    match channel {
        // One order per frame, at the top level rather than in a list.
        "orders_cross" => private_order(&value)
            .map(Event::OrderUpdate)
            .into_iter()
            .collect(),
        "accounts_cross" => {
            let empty = Vec::new();
            let rows = value
                .get("data")
                .and_then(serde_json::Value::as_array)
                .unwrap_or(&empty);
            let balances: Vec<Balance> = rows
                .iter()
                .filter_map(|row| {
                    Some(Balance {
                        asset: row.get("margin_asset")?.as_str()?.to_string(),
                        free: notification_decimal(row, "margin_available"),
                        // Margin committed to open orders and positions.
                        locked: notification_decimal(row, "margin_frozen"),
                    })
                })
                .collect();
            if balances.is_empty() {
                Vec::new()
            } else {
                vec![Event::BalanceUpdate(balances)]
            }
        }
        _ => Vec::new(),
    }
}

/// Build an [`Order`] from an `orders_cross` frame.
fn private_order(value: &serde_json::Value) -> Option<Order> {
    let contract = value.get("contract_code")?.as_str()?;
    let direction = value.get("direction")?.as_str()?;
    let price = notification_decimal(value, "price");
    let client_id = value.get("client_order_id").and_then(|id| {
        id.as_str()
            .map(str::to_string)
            .or_else(|| id.as_i64().map(|n| n.to_string()))
    });
    Some(Order {
        id: value
            .get("order_id_str")
            .and_then(serde_json::Value::as_str)
            .map(str::to_string)
            .or_else(|| {
                value
                    .get("order_id")
                    .and_then(serde_json::Value::as_i64)
                    .map(|n| n.to_string())
            })?,
        client_order_id: client_id,
        symbol: symbol_from_contract_code(contract),
        side: if direction == "sell" {
            OrderSide::Sell
        } else {
            OrderSide::Buy
        },
        // HTX names the kind by price type, the way its REST order does.
        order_type: if value
            .get("order_price_type")
            .and_then(serde_json::Value::as_str)
            == Some("limit")
        {
            OrderType::Limit
        } else {
            OrderType::Market
        },
        status: private_order_status(value.get("status").and_then(serde_json::Value::as_i64)),
        quantity: notification_decimal(value, "volume"),
        filled_quantity: notification_decimal(value, "trade_volume"),
        price: (price > Decimal::ZERO).then_some(price),
        average_price: {
            let average = notification_decimal(value, "trade_avg_price");
            (average > Decimal::ZERO).then_some(average)
        },
    })
}

/// Turn an HTX contract code back into a [`Symbol`].
///
/// `BTC-USDT` -> `BTC/USDT`. The private stream is not subscribed per symbol,
/// so there is no subscription to look the pair up in.
fn symbol_from_contract_code(code: &str) -> Symbol {
    code.split_once('-').map_or_else(
        || Symbol::new(code, ""),
        |(base, quote)| Symbol::new(base, quote),
    )
}

/// Map HTX's numeric futures order status to an [`OrderStatus`].
///
/// The venue numbers the whole lifecycle -- 1 preparing through 11 cancelling --
/// and only two of those numbers mean the order is done. A number this client
/// does not know is reported as still open rather than as filled, because
/// guessing "filled" ends a position in the caller's own bookkeeping.
fn private_order_status(status: Option<i64>) -> OrderStatus {
    match status {
        Some(4) => OrderStatus::PartiallyFilled,
        Some(5 | 7) => OrderStatus::Canceled,
        Some(6) => OrderStatus::Filled,
        _ => OrderStatus::New,
    }
}

/// Read a notification field HTX may send as a number or as a string.
fn notification_decimal(value: &serde_json::Value, key: &str) -> Decimal {
    value.get(key).map_or(Decimal::ZERO, |field| {
        field.as_str().map_or_else(
            || {
                field
                    .as_f64()
                    .and_then(|n| parse_decimal(&n.to_string()).ok())
                    .unwrap_or(Decimal::ZERO)
            },
            |text| parse_decimal(text).unwrap_or(Decimal::ZERO),
        )
    })
}

fn parse_notification(
    frame: &str,
    resolve: &impl Fn(&str) -> Symbol,
    subscribed: &impl Fn(&str, DerivativesChannel) -> bool,
) -> Vec<Event> {
    let Ok(value) = serde_json::from_str::<serde_json::Value>(frame) else {
        return Vec::new();
    };
    if value.get("op").and_then(serde_json::Value::as_str) != Some("notify") {
        return Vec::new();
    }
    let Some(topic) = value.get("topic").and_then(serde_json::Value::as_str) else {
        return Vec::new();
    };
    let mut parts = topic.split('.');
    let _public = parts.next();
    let code = parts.next().unwrap_or("");
    let channel = parts.next().unwrap_or("");
    let empty = Vec::new();
    let rows = value
        .get("data")
        .and_then(serde_json::Value::as_array)
        .unwrap_or(&empty);

    match channel {
        "funding_rate" if subscribed(code, DerivativesChannel::Funding) => rows
            .iter()
            .filter_map(|row| {
                Some(Event::Derivatives(DerivativesFeed::Funding(FundingRate {
                    symbol: resolve(row.get("contract_code")?.as_str()?),
                    rate: parse_decimal(row.get("funding_rate")?.as_str()?).ok()?,
                    // The funding frame carries no price; a mark taken from
                    // another channel would belong to another moment.
                    mark_price: Decimal::ZERO,
                    timestamp: row
                        .get("funding_time")
                        .and_then(serde_json::Value::as_str)
                        .and_then(|t| t.parse().ok())
                        .unwrap_or(0),
                })))
            })
            .collect(),
        "liquidation_orders" if subscribed(code, DerivativesChannel::Liquidations) => rows
            .iter()
            .filter_map(|row| {
                // `direction` is the side of the forced order hitting the book:
                // a liquidated long is sold. `volume` is the contract count,
                // the unit this client's swap stream uses.
                let price = row.get("price").and_then(serde_json::Value::as_f64)?;
                let volume = row.get("volume").and_then(serde_json::Value::as_f64)?;
                Some(Event::Derivatives(DerivativesFeed::Liquidation(
                    Liquidation {
                        symbol: resolve(row.get("contract_code")?.as_str()?),
                        side: parse_side(row.get("direction")?.as_str()?).ok()?,
                        price: parse_decimal(&price.to_string()).ok()?,
                        quantity: parse_decimal(&volume.to_string()).ok()?,
                        timestamp: row
                            .get("created_at")
                            .and_then(serde_json::Value::as_i64)
                            .unwrap_or(0),
                    },
                )))
            })
            .collect(),
        _ => Vec::new(),
    }
}

/// The reply HTX's heartbeat asks for, or `None` if this frame is not one.
///
/// Two shapes, because HTX runs two protocols: the public market socket sends
/// `{"ping": <ms>}` and the private v2 socket `{"action":"ping","data":{"ts":..}}`.
/// Both want the value echoed back. A frame that goes unanswered closes the
/// stream -- and since the reconnect opens another socket that is closed for the
/// same reason, the result is a stream that reconnects forever and delivers
/// nothing.
fn pong_for(frame: &str) -> Option<String> {
    let value: serde_json::Value = serde_json::from_str(frame).ok()?;
    if let Some(ping) = value.get("ping") {
        return Some(format!(r#"{{"pong":{ping}}}"#));
    }
    // The notification socket -- public derivatives *and* the futures private
    // stream -- pings every five seconds in a third form, and closes the socket
    // when it is not answered. `ts` comes back as a string there and must be
    // echoed as it arrived.
    if value.get("op").and_then(serde_json::Value::as_str) == Some("ping") {
        let ts = value.get("ts").cloned().unwrap_or(serde_json::Value::Null);
        return Some(format!(r#"{{"op":"pong","ts":{ts}}}"#));
    }
    if value.get("action").and_then(serde_json::Value::as_str) == Some("ping") {
        let ts = value
            .get("data")
            .and_then(|d| d.get("ts"))
            .cloned()
            .unwrap_or(serde_json::Value::Null);
        return Some(format!(r#"{{"action":"pong","data":{{"ts":{ts}}}}}"#));
    }
    None
}

fn parse_ws_message(text: &str, resolve: &impl Fn(&str) -> Symbol) -> Result<Vec<Event>> {
    let value: serde_json::Value =
        serde_json::from_str(text).map_err(|e| Error::Deserialization(e.to_string()))?;
    // HTX v2 private stream frames are `{"action":"push","ch":"orders#..."|"accounts.update#...","data":{...}}`.
    if value.get("action").and_then(serde_json::Value::as_str) == Some("push") {
        let ch = value
            .get("ch")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("");
        let null = serde_json::Value::Null;
        let data = value.get("data").unwrap_or(&null);
        if ch.starts_with("orders#") {
            return Ok(vec![Event::OrderUpdate(ws_order_from_data(data)?)]);
        } else if ch.starts_with("accounts.update") {
            return Ok(vec![Event::BalanceUpdate(vec![ws_balance_from_data(
                data,
            )?])]);
        }
        return Ok(Vec::new());
    }
    let Some(channel) = value.get("ch").and_then(serde_json::Value::as_str) else {
        return Ok(Vec::new()); // ping / sub-ack
    };
    // channel = "market.<symbol>.<stream>...".
    let mut parts = channel.split('.');
    let _market = parts.next();
    let wire = parts.next().unwrap_or("");
    let stream = parts.next().unwrap_or("");
    let symbol = resolve(wire);
    let null = serde_json::Value::Null;
    let tick = value.get("tick").unwrap_or(&null);

    match stream {
        "trade" => {
            let data = tick.get("data").and_then(serde_json::Value::as_array);
            let empty = Vec::new();
            data.unwrap_or(&empty)
                .iter()
                .map(|t| {
                    let side = t
                        .get("direction")
                        .and_then(serde_json::Value::as_str)
                        .unwrap_or("");
                    Ok(Event::Trade(TradePrint {
                        symbol: symbol.clone(),
                        price: decimal_field(t, "price")?,
                        quantity: decimal_field(t, "amount")?,
                        aggressor: parse_side(side)?,
                        timestamp: t.get("ts").and_then(serde_json::Value::as_i64).unwrap_or(0),
                    }))
                })
                .collect()
        }
        "ticker" => Ok(vec![Event::Ticker(Ticker {
            symbol,
            last: decimal_field(tick, "lastPrice").or_else(|_| decimal_field(tick, "close"))?,
            bid: decimal_field(tick, "bid").unwrap_or(Decimal::ZERO),
            ask: decimal_field(tick, "ask").unwrap_or(Decimal::ZERO),
            volume: decimal_field(tick, "vol").unwrap_or(Decimal::ZERO),
            timestamp: value
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })]),
        // The futures quote channel. Its top of book arrives as `[price, size]`
        // pairs rather than scalars, so the spot arm's `bid` / `ask` reads would
        // silently fall back to zero here.
        "detail" => Ok(vec![Event::Ticker(Ticker {
            symbol,
            last: decimal_field(tick, "close")?,
            bid: level_price(tick, "bid"),
            ask: level_price(tick, "ask"),
            volume: decimal_field(tick, "amount").unwrap_or(Decimal::ZERO),
            timestamp: value
                .get("ts")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })]),
        "depth" => {
            let update_id = tick
                .get("version")
                .and_then(serde_json::Value::as_u64)
                .unwrap_or(0);
            Ok(vec![Event::BookDelta(BookDelta {
                symbol,
                first_update_id: update_id,
                final_update_id: update_id,
                bids: num_levels(tick.get("bids"))?,
                asks: num_levels(tick.get("asks"))?,
                timestamp: tick
                    .get("ts")
                    .and_then(serde_json::Value::as_i64)
                    .unwrap_or(0),
            })])
        }
        _ => Ok(Vec::new()),
    }
}

impl MarketData for Htx {
    fn ticker(&mut self, symbol: &Symbol) -> Result<Ticker> {
        Htx::ticker(self, symbol)
    }
    fn klines(&mut self, symbol: &Symbol, interval: &str, limit: u32) -> Result<Vec<Candle>> {
        Htx::klines(self, symbol, interval, limit)
    }
    fn order_book(&mut self, symbol: &Symbol, depth: u32) -> Result<OrderBookSnapshot> {
        Htx::order_book(self, symbol, depth)
    }
    fn subscribe_trades(&mut self, symbol: &Symbol) -> Result<()> {
        Htx::subscribe_trades(self, symbol)
    }
    fn subscribe_book(&mut self, symbol: &Symbol) -> Result<()> {
        Htx::subscribe_book(self, symbol)
    }
    fn subscribe_ticker(&mut self, symbol: &Symbol) -> Result<()> {
        Htx::subscribe_ticker(self, symbol)
    }
    fn poll_events(&mut self) -> Vec<Event> {
        Htx::poll_events(self)
    }
}

impl Execution for Htx {
    fn place_order(&mut self, request: &OrderRequest) -> Result<Order> {
        Htx::place_order(self, request)
    }
    fn cancel_order(&mut self, symbol: &Symbol, order_id: &str) -> Result<()> {
        Htx::cancel_order(self, symbol, order_id)
    }
    fn query_order(&mut self, symbol: &Symbol, order_id: &str) -> Result<Order> {
        Htx::query_order(self, symbol, order_id)
    }
    fn open_orders(&mut self, symbol: Option<&Symbol>) -> Result<Vec<Order>> {
        Htx::open_orders(self, symbol)
    }
    fn balances(&mut self) -> Result<Vec<Balance>> {
        Htx::balances(self)
    }
}

impl Exchange for Htx {
    fn name(&self) -> &'static str {
        "htx"
    }
}

impl WsUserData for Htx {
    fn subscribe_user_data(&mut self) -> Result<()> {
        Htx::subscribe_user_data(self)
    }
    fn keepalive_user_data(&mut self) -> Result<()> {
        Htx::keepalive_user_data(self)
    }
}

impl WsExecution for Htx {
    /// HTX exposes no WebSocket order-entry API for spot — its WebSocket surface
    /// is subscription-only (market data plus the private v2 order/account push
    /// channels). Orders are placed over REST, so this returns a documented
    /// [`Error::Exchange`].
    fn place_order_ws(&mut self, _request: &OrderRequest) -> Result<Order> {
        Err(Error::Exchange {
            code: "unsupported".to_string(),
            message: "HTX has no WebSocket order-entry API; place orders over REST \
                      (POST /v1/order/orders/place)"
                .to_string(),
        })
    }

    fn cancel_order_ws(&mut self, _symbol: &Symbol, _order_id: &str) -> Result<()> {
        Err(Error::Exchange {
            code: "unsupported".to_string(),
            message: "HTX has no WebSocket order-entry API; cancel orders over REST \
                      (POST /v1/order/orders/{id}/submitcancel)"
                .to_string(),
        })
    }
}

impl Derivatives for Htx {
    fn positions(&mut self, symbol: Option<&Symbol>) -> Result<Vec<Position>> {
        Htx::positions(self, symbol)
    }
    fn set_leverage(&mut self, symbol: &Symbol, leverage: u32) -> Result<()> {
        Htx::set_leverage(self, symbol, leverage)
    }
    fn set_margin_mode(&mut self, symbol: &Symbol, mode: MarginMode) -> Result<()> {
        Htx::set_margin_mode(self, symbol, mode)
    }
    fn close_position(&mut self, symbol: &Symbol) -> Result<Order> {
        Htx::close_position(self, symbol)
    }
}

impl DerivativesStream for Htx {
    fn subscribe_derivatives(
        &mut self,
        symbol: &Symbol,
        channel: DerivativesChannel,
    ) -> Result<()> {
        Htx::subscribe_derivatives(self, symbol, channel)
    }
    fn open_interest(&mut self, symbol: &Symbol) -> Result<OpenInterest> {
        Htx::open_interest(self, symbol)
    }
    fn long_short_ratio(&mut self, symbol: &Symbol) -> Result<LongShortRatio> {
        Htx::long_short_ratio(self, symbol)
    }
}

impl Htx {
    /// Place several spot orders in one request (`/v1/order/batch-orders`). Each
    /// element carries an `err-code`, so each request's own [`Result`] is kept.
    ///
    /// # Errors
    /// Returns an [`Error`] if the batch request itself fails, or if called on a
    /// futures client (spot batch endpoint).
    pub fn place_batch(&self, requests: &[OrderRequest]) -> Result<Vec<Result<Order>>> {
        if requests.iter().any(|r| r.order_type.is_trigger()) {
            return Err(Error::unsupported_trigger("HTX"));
        }
        for request in requests {
            self.ensure_reduce_only_is_reducible(request)?;
        }
        if requests.is_empty() {
            return Ok(Vec::new());
        }
        if self.is_futures() {
            return Err(Error::Exchange {
                code: "unsupported".to_string(),
                message: "HTX batch-orders is a spot endpoint".to_string(),
            });
        }
        // Resolved up front, so a request this endpoint cannot express refuses
        // the whole batch rather than travelling with a field dropped -- and
        // before the account lookup, so nothing reaches the wire either.
        for r in requests {
            ensure_no_stp(r)?;
        }
        let kinds = requests
            .iter()
            .map(order_kind_str)
            .collect::<Result<Vec<_>>>()?;
        let account_id = self.account_id()?;
        let items: Vec<serde_json::Value> = requests
            .iter()
            .zip(&kinds)
            .map(|(r, kind)| {
                let order_kind = format!("{}-{}", side_str(r.side), kind);
                let mut o = serde_json::json!({
                    "account-id": account_id,
                    "symbol": Self::wire_symbol(&r.symbol),
                    "type": order_kind,
                    "amount": format_decimal(r.quantity),
                });
                if let Some(price) = r.price {
                    o["price"] = serde_json::json!(format_decimal(price));
                }
                if let Some(id) = &r.client_order_id {
                    o["client-order-id"] = serde_json::json!(id.clone());
                }
                o
            })
            .collect();
        let body = serde_json::Value::Array(items).to_string();
        let value = self.signed_request(HttpMethod::Post, "/v1/order/batch-orders", &[], &body)?;
        let data = value
            .get("data")
            .and_then(serde_json::Value::as_array)
            .ok_or_else(|| Error::Deserialization("missing batch data".to_string()))?;
        Ok(requests
            .iter()
            .zip(data)
            .map(|(req, elem)| {
                let err_code = elem
                    .get("err-code")
                    .and_then(serde_json::Value::as_str)
                    .unwrap_or("");
                if !err_code.is_empty() {
                    return Err(Error::OrderRejected {
                        code: err_code.to_string(),
                        message: elem
                            .get("err-msg")
                            .and_then(serde_json::Value::as_str)
                            .unwrap_or("")
                            .to_string(),
                    });
                }
                let id = elem
                    .get("order-id")
                    .map(|v| {
                        v.as_u64()
                            .map_or_else(|| v.as_str().unwrap_or("").to_string(), |n| n.to_string())
                    })
                    .unwrap_or_default();
                Ok(Order {
                    id,
                    client_order_id: req.client_order_id.clone(),
                    symbol: req.symbol.clone(),
                    side: req.side,
                    order_type: req.order_type,
                    status: OrderStatus::New,
                    quantity: req.quantity,
                    filled_quantity: Decimal::ZERO,
                    price: req.price,
                    average_price: None,
                })
            })
            .collect())
    }

    /// Cancel several orders by id in one request (`/v1/order/orders/batchcancel`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails.
    pub fn cancel_batch(&self, _symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        let ids: Vec<&str> = order_ids.iter().map(String::as_str).collect();
        let body = serde_json::json!({ "order-ids": ids }).to_string();
        self.signed_request(HttpMethod::Post, "/v1/order/orders/batchcancel", &[], &body)?;
        Ok(())
    }
}

impl AdvancedOrders for Htx {
    /// HTX has no in-place amend (orders are cancelled and re-placed), so this
    /// returns an [`Error::Exchange`].
    fn amend_order(
        &mut self,
        _symbol: &Symbol,
        _order_id: &str,
        _new_price: Option<Decimal>,
        _new_quantity: Option<Decimal>,
    ) -> Result<Order> {
        Err(Error::Exchange {
            code: "unsupported".to_string(),
            message: "HTX has no in-place amend; cancel and re-place the order".to_string(),
        })
    }
    fn place_batch(&mut self, requests: &[OrderRequest]) -> Result<Vec<Result<Order>>> {
        Htx::place_batch(self, requests)
    }
    fn cancel_batch(&mut self, symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        Htx::cancel_batch(self, symbol, order_ids)
    }
    /// HTX has no OCO order-list, so this returns an [`Error::Exchange`].
    fn place_oco(&mut self, _request: &OcoRequest) -> Result<Vec<Order>> {
        Err(Error::Exchange {
            code: "unsupported".to_string(),
            message: "HTX has no OCO order-list".to_string(),
        })
    }
}

/// Round a base quantity to a whole number of HTX contracts.
fn decimal_to_contracts(quantity: Decimal) -> Result<i64> {
    quantity
        .round()
        .to_i64()
        .filter(|c| *c > 0)
        .ok_or(Error::InvalidOrder(
            "futures volume rounds to zero contracts",
        ))
}

fn parse_futures_position(data: &serde_json::Value) -> Result<Position> {
    let direction = data
        .get("direction")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let side = match direction {
        "buy" => PositionSide::Long,
        "sell" => PositionSide::Short,
        other => {
            return Err(Error::Deserialization(format!(
                "unknown position direction {other:?}"
            )))
        }
    };
    let contract = data
        .get("contract_code")
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    Ok(Position {
        symbol: symbol_from_contract(contract),
        side,
        quantity: decimal_field(data, "volume").unwrap_or(Decimal::ZERO),
        entry_price: decimal_field(data, "cost_open").unwrap_or(Decimal::ZERO),
        mark_price: decimal_field(data, "last_price").unwrap_or(Decimal::ZERO),
        leverage: decimal_field(data, "lever_rate").unwrap_or(Decimal::ZERO),
        unrealized_pnl: decimal_field(data, "profit_unreal").unwrap_or(Decimal::ZERO),
        // The cross-margin family always reports cross positions.
        margin_mode: MarginMode::Cross,
    })
}

/// Reconstruct a canonical [`Symbol`] from an HTX swap contract code
/// (`BTC-USDT` -> `BTC/USDT`).
fn symbol_from_contract(contract: &str) -> Symbol {
    match contract.split_once('-') {
        Some((base, quote)) => Symbol::new(base, quote),
        None => Symbol::new(contract, ""),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::{HttpResponse, MockHttpTransport, MockWsTransport};
    use rust_decimal_macros::dec;
    use std::sync::Arc;

    struct ArcTransport(Arc<MockHttpTransport>);
    impl HttpTransport for ArcTransport {
        fn execute(&self, request: &HttpRequest) -> Result<HttpResponse> {
            self.0.execute(request)
        }
    }
    struct ArcWs(Arc<MockWsTransport>);
    impl WsTransport for ArcWs {
        fn connect(&self, url: &str) -> Result<Box<dyn WsConnection>> {
            self.0.connect(url)
        }
    }

    fn symbol() -> Symbol {
        Symbol::new("BTC", "USDT")
    }

    #[test]
    fn a_blocked_reply_is_reported_as_the_block_not_as_drift() {
        let (htx, mock) = client();
        // What a geo-blocked runner receives: a status of its own and a body
        // that was never this venue's JSON. Read as an envelope it fails at
        // line 2 column 5, which says nothing about why it failed.
        mock.push_json(403, "{\n    error: 'unavailable in your region'\n}");
        let error = htx.ticker(&symbol()).unwrap_err();
        let named = matches!(&error, Error::Exchange { code, .. } if code == "403");
        assert!(named, "a blocked reply must not be reported as drift");
    }

    fn client() -> (Htx, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        (
            Htx::with_http(Box::new(ArcTransport(Arc::clone(&mock))), &opts),
            mock,
        )
    }

    fn signed_client(now_ms: i64) -> (Htx, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let htx = Htx::with_credentials(
            Box::new(ArcTransport(Arc::clone(&mock))),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_clock(Box::new(move || now_ms));
        (htx, mock)
    }

    fn futures_client() -> (Htx, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        (
            Htx::with_http(Box::new(ArcTransport(Arc::clone(&mock))), &opts),
            mock,
        )
    }

    fn signed_futures_client(now_ms: i64) -> (Htx, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_credentials(
            Box::new(ArcTransport(Arc::clone(&mock))),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_clock(Box::new(move || now_ms));
        (htx, mock)
    }

    fn signed_futures_ws_client(now_ms: i64) -> (Htx, Arc<MockWsTransport>) {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_credentials(
            Box::new(ArcTransport(http)),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))))
        .with_clock(Box::new(move || now_ms));
        (htx, ws)
    }

    /// A futures client watches the futures account, on the notification socket.
    ///
    /// Verified against the live venue: that socket answers
    /// `{"op":"sub","topic":"orders_cross.*"}` with `err-code 2002,
    /// Authentication required.` -- an unrecognised topic answers differently.
    #[test]
    fn a_futures_client_authenticates_and_subscribes_to_the_cross_topics() {
        let (mut htx, ws) = signed_futures_ws_client(1_700_000_000_000);
        htx.subscribe_user_data().unwrap();

        assert_eq!(
            ws.connected_urls()[0],
            "wss://api.hbdm.com/linear-swap-notification"
        );
        let sent = ws.sent();
        assert!(sent[0].contains(r#""op":"auth""#), "{}", sent[0]);
        assert!(sent[0].contains(r#""AccessKeyId":"APIKEY""#), "{}", sent[0]);
        assert!(
            sent[1].contains(r#""topic":"orders_cross.*""#),
            "{}",
            sent[1]
        );
        assert!(
            sent[2].contains(r#""topic":"accounts_cross.*""#),
            "{}",
            sent[2]
        );
        // The spot v2 stream is a different host and a different envelope.
        assert!(
            !sent.iter().any(|frame| frame.contains("orders#")),
            "a futures client subscribed to the spot stream: {sent:?}"
        );
    }

    /// The notification socket pings in a form neither of the other two matches,
    /// every five seconds, and closes the socket when it goes unanswered --
    /// which is what the public derivatives channels were doing.
    ///
    /// `ts` arrives as a string here and has to be echoed as it came.
    #[test]
    fn the_notification_ping_is_answered_in_its_own_form() {
        assert_eq!(
            pong_for(r#"{"op":"ping","ts":"1788470774457"}"#).unwrap(),
            r#"{"op":"pong","ts":"1788470774457"}"#
        );
        // The other two forms still answer as they did.
        assert_eq!(pong_for(r#"{"ping":123}"#).unwrap(), r#"{"pong":123}"#);
        assert!(pong_for(r#"{"action":"ping","data":{"ts":9}}"#).is_some());
        assert!(pong_for(r#"{"op":"notify","topic":"x"}"#).is_none());
    }

    #[test]
    fn a_cross_order_frame_becomes_an_order_update() {
        let (mut htx, ws) = signed_futures_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"op":"notify","topic":"orders_cross.btc-usdt","ts":1700000000000,
                "symbol":"BTC","contract_code":"BTC-USDT","volume":4,"price":19000,
                "order_price_type":"limit","direction":"sell","offset":"open","status":4,
                "order_id_str":"9911","client_order_id":77,"trade_volume":3,
                "trade_avg_price":19010}"#
                .to_string(),
        ))]);
        htx.subscribe_user_data().unwrap();

        let events = htx.poll_events();
        let Some(Event::OrderUpdate(order)) = events.first() else {
            panic!("expected an order update, got {events:?}");
        };
        assert_eq!(order.id, "9911");
        assert_eq!(order.symbol, Symbol::new("BTC", "USDT"));
        assert_eq!(order.side, OrderSide::Sell);
        // Status 4 is a partial fill; the venue numbers the whole lifecycle.
        assert_eq!(order.status, OrderStatus::PartiallyFilled);
        assert_eq!(order.quantity, dec!(4));
        assert_eq!(order.filled_quantity, dec!(3));
        assert_eq!(order.client_order_id.as_deref(), Some("77"));
        assert_eq!(order.average_price, Some(dec!(19010)));
    }

    /// HTX numbers the order lifecycle, and only two of those numbers mean the
    /// order is done. An unknown number is reported as still open: guessing
    /// "filled" would close a position in the caller's own bookkeeping that the
    /// venue still has open.
    #[test]
    fn an_unknown_order_status_number_is_not_reported_as_filled() {
        assert_eq!(private_order_status(Some(6)), OrderStatus::Filled);
        assert_eq!(private_order_status(Some(7)), OrderStatus::Canceled);
        assert_eq!(private_order_status(Some(5)), OrderStatus::Canceled);
        assert_eq!(private_order_status(Some(3)), OrderStatus::New);
        assert_eq!(private_order_status(Some(99)), OrderStatus::New);
        assert_eq!(private_order_status(None), OrderStatus::New);
    }

    #[test]
    fn a_cross_account_frame_becomes_a_balance_update() {
        let (mut htx, ws) = signed_futures_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"op":"notify","topic":"accounts_cross.USDT","ts":1700000000000,
                "data":[{"margin_asset":"USDT","margin_balance":1000.5,
                         "margin_available":900.25,"margin_frozen":100.25}]}"#
                .to_string(),
        ))]);
        htx.subscribe_user_data().unwrap();

        let events = htx.poll_events();
        let Some(Event::BalanceUpdate(balances)) = events.first() else {
            panic!("expected a balance update, got {events:?}");
        };
        assert_eq!(balances[0].asset, "USDT");
        assert_eq!(balances[0].free, dec!(900.25));
        // Margin committed to open orders and positions is locked, not free.
        assert_eq!(balances[0].locked, dec!(100.25));
    }

    /// A spot trigger goes to the algo endpoint, which is a different path than
    /// an ordinary spot order and a different envelope in the answer.
    ///
    /// It is addressed by its client order id, so one is always sent -- an
    /// unnamed algo order could not be cancelled afterwards. There is no
    /// comparison field here: spot reads the direction off the side.
    #[test]
    fn a_spot_trigger_goes_to_the_algo_endpoint_and_is_named() {
        let (htx, mock) = signed_client(1_700_000_000_000);
        // The account lookup first: a spot algo order carries the account id.
        mock.push_json(200, r#"{"status":"ok","data":[{"id":42,"type":"spot"}]}"#);
        // The v2 envelope, which says `code`, not `status`.
        mock.push_json(200, r#"{"code":200,"data":{"clientOrderId":"a-1"}}"#);
        let request = OrderRequest {
            order_type: OrderType::StopMarket,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::market_sell(symbol(), dec!(1))
        };
        let order = htx.place_order(&request).unwrap();
        assert_eq!(order.id, "a-1");

        let sent = mock.recorded_requests();
        let placed = sent.last().unwrap();
        assert!(placed.url.contains("/v2/algo-orders"), "{}", placed.url);
        let body = placed.body.clone().unwrap();
        assert!(body.contains(r#""stopPrice":"19000""#), "{body}");
        assert!(body.contains(r#""accountId":"42""#), "{body}");
        assert!(body.contains(r#""orderSide":"sell""#), "{body}");
        assert!(body.contains(r#""orderType":"market""#), "{body}");
        assert!(body.contains(r#""clientOrderId":"wkalgo"#), "{body}");
        // A market trigger has no limit price, and an empty one is not a price.
        assert!(!body.contains("orderPrice"), "{body}");
    }

    /// The v2 endpoints answer `code: 200` and carry no `status` at all, so a
    /// client that reads only the v1 envelope reports every placed algo order as
    /// a failure with an empty code and an empty message.
    #[test]
    fn the_v2_envelope_is_read_as_success_and_its_errors_as_errors() {
        assert!(unwrap_status(r#"{"code":200,"data":{"clientOrderId":"a"}}"#).is_ok());
        let err = unwrap_status(r#"{"code":2002,"message":"invalid parameter"}"#).unwrap_err();
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "2002"));
    }

    /// The swap trigger goes to its own endpoint, and spells "take the market"
    /// as an order price type rather than as a price.
    #[test]
    fn a_swap_trigger_goes_to_the_trigger_endpoint_and_names_its_comparison() {
        let (htx, mock) = signed_futures_client(1_700_000_000_000);
        mock.push_json(200, r#"{"status":"ok","data":{"order_id":7}}"#);
        let request = OrderRequest {
            order_type: OrderType::StopMarket,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::market_sell(symbol(), dec!(1))
        };
        htx.place_order(&request).unwrap();

        let sent = &mock.recorded_requests()[0];
        assert!(sent.url.contains("swap_trigger_order"), "{}", sent.url);
        let body = sent.body.clone().unwrap();
        assert!(body.contains(r#""trigger_price":"19000""#), "{body}");
        // Selling to protect a long fires on the way down.
        assert!(body.contains(r#""trigger_type":"le""#), "{body}");
        // No price, so the order takes the market -- said as a price *type*.
        assert!(body.contains(r#""order_price_type":"optimal_5""#), "{body}");
    }

    /// A limit trigger sends its limit price, and says so.
    #[test]
    fn a_swap_stop_limit_carries_its_limit_price() {
        let (htx, mock) = signed_futures_client(1_700_000_000_000);
        mock.push_json(200, r#"{"status":"ok","data":{"order_id":8}}"#);
        let request = OrderRequest {
            order_type: OrderType::StopLimit,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::limit_sell(symbol(), dec!(1), dec!(18900))
        };
        htx.place_order(&request).unwrap();

        let body = mock.recorded_requests()[0].body.clone().unwrap();
        assert!(body.contains(r#""order_price_type":"limit""#), "{body}");
        assert!(body.contains(r#""order_price":"18900""#), "{body}");
    }

    #[test]
    fn an_htx_trigger_without_a_stop_price_is_refused() {
        let (htx, mock) = signed_futures_client(1_700_000_000_000);
        let request = OrderRequest {
            order_type: OrderType::StopMarket,
            stop_price: None,
            ..OrderRequest::market_sell(symbol(), dec!(1))
        };
        assert!(htx.place_order(&request).is_err());
        assert_eq!(mock.recorded_requests(), Vec::new());
    }

    fn signed_ws_client(now_ms: i64) -> (Htx, Arc<MockWsTransport>) {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let htx = Htx::with_credentials(
            Box::new(ArcTransport(http)),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))))
        .with_clock(Box::new(move || now_ms));
        (htx, ws)
    }

    fn futures_ws_client() -> (Htx, Arc<MockWsTransport>) {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        (htx, ws)
    }

    /// The answers HTX gives on a bad day.
    #[test]
    fn htx_polled_reads_report_what_went_wrong() {
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_http(Box::new(ArcTransport(Arc::clone(&http))), &opts);

        http.push_json(200, r#"{"status":"ok","data":[],"ts":1}"#);
        assert!(matches!(
            Htx::open_interest(&htx, &symbol()),
            Err(Error::NotFound(_))
        ));

        http.push_json(
            200,
            r#"{"status":"ok","data":[{"contract_code":"BTC-USDT"}],"ts":1}"#,
        );
        assert!(matches!(
            Htx::open_interest(&htx, &symbol()),
            Err(Error::Deserialization(_))
        ));

        http.push_json(200, r#"{"status":"ok","data":{"list":[]},"ts":1}"#);
        assert!(matches!(
            Htx::long_short_ratio(&htx, &symbol()),
            Err(Error::NotFound(_))
        ));

        http.push_json(
            200,
            r#"{"status":"ok","data":{"list":[{"sell_ratio":0.2,"ts":1}]},"ts":1}"#,
        );
        assert!(matches!(
            Htx::long_short_ratio(&htx, &symbol()),
            Err(Error::Deserialization(_))
        ));
    }

    /// Notification frames the parser cannot use are dropped.
    #[test]
    fn htx_drops_notification_frames_it_cannot_read() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![
            Ok(Some("not json".to_string())),
            // A notify with no topic.
            Ok(Some(r#"{"op":"notify","data":[]}"#.to_string())),
            // A channel this client does not subscribe to.
            Ok(Some(
                r#"{"op":"notify","topic":"public.BTC-USDT.something","data":[]}"#.to_string(),
            )),
            // A funding row with no rate.
            Ok(Some(
                r#"{"op":"notify","topic":"public.BTC-USDT.funding_rate",
                "data":[{"contract_code":"BTC-USDT"}]}"#
                    .to_string(),
            )),
            // A liquidation row with no price.
            Ok(Some(
                r#"{"op":"notify","topic":"public.BTC-USDT.liquidation_orders",
                "data":[{"contract_code":"BTC-USDT","direction":"sell","volume":1}]}"#
                    .to_string(),
            )),
        ]);
        Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::Funding).unwrap();
        Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::Liquidations).unwrap();

        assert_eq!(Htx::poll_events(&mut htx), Vec::new());
    }

    /// The futures quote channel without its paired top of book.
    #[test]
    fn htx_detail_without_pairs_reports_no_top_of_book() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"ch":"market.BTC-USDT.detail","ts":1,"tick":{"close":81217.7}}"#.to_string(),
        ))]);
        Htx::subscribe_ticker(&mut htx, &symbol()).unwrap();

        let events = Htx::poll_events(&mut htx);
        let Event::Ticker(ticker) = &events[0] else {
            panic!("expected a ticker, got {:?}", events[0]);
        };
        assert_eq!(ticker.bid, Decimal::ZERO);
        assert_eq!(ticker.ask, Decimal::ZERO);
    }

    /// Funding and forced orders arrive on a second public socket.
    #[test]
    fn the_notification_socket_carries_funding_and_liquidations() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![
            Ok(Some(
                r#"{"op":"notify","topic":"public.BTC-USDT.funding_rate","ts":1788465096842,
                "data":[{"symbol":"BTC","contract_code":"BTC-USDT","fee_asset":"USDT",
                "funding_time":"1788465090000","funding_rate":"0.000035251752528159",
                "estimated_rate":null,"settlement_time":"1788480000000"}]}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"op":"notify","topic":"public.BTC-USDT.liquidation_orders","ts":1788465321721,
                "data":[{"symbol":"BTC","contract_code":"BTC-USDT","direction":"sell",
                "offset":"close","volume":4,"price":81894.5,"created_at":1788465321715,
                "amount":0.004,"trade_turnover":327.578}]}"#
                    .to_string(),
            )),
        ]);
        Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::Funding).unwrap();
        Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::Liquidations).unwrap();

        assert_eq!(
            ws.connected_urls()[0],
            "wss://api.hbdm.com/linear-swap-notification"
        );
        assert!(ws.sent()[0].contains("public.BTC-USDT.funding_rate"));

        let events = Htx::poll_events(&mut htx);
        let Event::Derivatives(DerivativesFeed::Funding(funding)) = &events[0] else {
            panic!("expected funding, got {:?}", events[0]);
        };
        assert_eq!(funding.rate, dec!(0.000035251752528159));
        // The funding frame carries no price; a mark from another channel would
        // belong to another moment.
        assert_eq!(funding.mark_price, Decimal::ZERO);
        assert_eq!(funding.timestamp, 1_788_465_090_000);

        let Event::Derivatives(DerivativesFeed::Liquidation(liquidation)) = &events[1] else {
            panic!("expected a liquidation, got {:?}", events[1]);
        };
        // A liquidated long is sold, which is the side of the forced order.
        assert_eq!(liquidation.side, OrderSide::Sell);
        assert_eq!(liquidation.price, dec!(81894.5));
        // `volume` is the contract count, the unit the swap stream uses.
        assert_eq!(liquidation.quantity, dec!(4));
        assert_eq!(liquidation.timestamp, 1_788_465_321_715);
    }

    /// A subscribe acknowledgement is not an empty channel.
    #[test]
    fn a_subscribe_ack_on_the_notification_socket_is_not_an_event() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"op":"sub","cid":"BTC-USDT","topic":"public.BTC-USDT.liquidation_orders",
            "ts":1788465096842,"err-code":0}"#
                .to_string(),
        ))]);
        Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::Liquidations).unwrap();

        assert_eq!(Htx::poll_events(&mut htx), Vec::new());
    }

    /// HTX publishes mark and index as separate kline channels, never together.
    #[test]
    fn htx_refuses_a_combined_mark_index_channel() {
        let (mut htx, ws) = futures_ws_client();
        let err = Htx::subscribe_derivatives(&mut htx, &symbol(), DerivativesChannel::MarkIndex)
            .expect_err("HTX publishes no combined mark/index frame");
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert_eq!(ws.sent(), Vec::<String>::new());
    }

    /// HTX's field names invert between endpoints: here `volume` is contracts
    /// and `amount` is base currency, while on the trade stream `amount` is the
    /// contract count. Taking `amount` would silently switch units.
    #[test]
    fn open_interest_reads_the_contract_count_not_the_base_amount() {
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_http(Box::new(ArcTransport(Arc::clone(&http))), &opts);
        http.push_json(
            200,
            r#"{"status":"ok","data":[{"volume":29406469,"amount":29406.469,"symbol":"BTC",
            "value":2397165361.88,"contract_code":"BTC-USDT"}],"ts":1788464583675}"#,
        );
        let open_interest = Htx::open_interest(&htx, &symbol()).unwrap();
        assert_eq!(open_interest.open_interest, dec!(29406469));
        assert_eq!(open_interest.timestamp, 1_788_464_583_675);
    }

    /// The elite ratio's two figures do not sum to one: `locked_ratio` holds the
    /// remainder. They are carried as given rather than rescaled, since
    /// rescaling would invent proportions the venue did not publish.
    #[test]
    fn the_elite_ratio_is_carried_as_published() {
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let htx = Htx::with_http(Box::new(ArcTransport(Arc::clone(&http))), &opts);
        http.push_json(
            200,
            r#"{"status":"ok","data":{"list":[
            {"buy_ratio":0.51,"sell_ratio":0.27,"locked_ratio":0.22,"ts":1788455700000},
            {"buy_ratio":0.53,"sell_ratio":0.26,"locked_ratio":0.21,"ts":1788456000000}]},
            "ts":1788464583675}"#,
        );
        let ratio = Htx::long_short_ratio(&htx, &symbol()).unwrap();
        // Oldest first here, so the newest is the last row.
        assert_eq!(ratio.long_size, dec!(0.53));
        assert_eq!(ratio.short_size, dec!(0.26));
        assert_ne!(ratio.long_size + ratio.short_size, Decimal::ONE);
        assert_eq!(ratio.timestamp, 1_788_456_000_000);
    }

    #[test]
    fn a_spot_client_refuses_the_htx_derivatives_surface() {
        let ws = Arc::new(MockWsTransport::new());
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut htx = Htx::with_http(Box::new(ArcTransport(Arc::clone(&http))), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        for channel in [
            DerivativesChannel::Funding,
            DerivativesChannel::MarkIndex,
            DerivativesChannel::Liquidations,
        ] {
            assert!(Htx::subscribe_derivatives(&mut htx, &symbol(), channel).is_err());
        }
        assert!(Htx::open_interest(&htx, &symbol()).is_err());
        assert!(Htx::long_short_ratio(&htx, &symbol()).is_err());
    }

    /// Host, symbol form and quote channel all follow the market.
    #[test]
    fn a_futures_client_subscribes_on_the_linear_swap_host() {
        let (mut htx, ws) = futures_ws_client();
        Htx::subscribe_trades(&mut htx, &symbol()).unwrap();
        Htx::subscribe_ticker(&mut htx, &symbol()).unwrap();

        assert_eq!(ws.connected_urls()[0], "wss://api.hbdm.com/linear-swap-ws");
        assert!(ws.sent()[0].contains("market.BTC-USDT.trade.detail"));
        // The futures quote channel is `detail`, not `ticker`.
        assert!(ws.sent()[1].contains("market.BTC-USDT.detail"));
        assert!(ws.sent().iter().all(|f| !f.contains("btcusdt")));
    }

    /// A real linear-swap trade frame. `amount` is the contract count, which is
    /// the unit the futures order path takes, so the arm needs no special case.
    #[test]
    fn a_futures_trade_frame_parses() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"ch":"market.BTC-USDT.trade.detail","ts":1788462493057,"tick":{
            "id":100127722487328,"ts":1788462493055,"data":[{"amount":2,"quantity":0.002,
            "trade_turnover":162.4354,"ts":1788462493055,"id":1001277224873280000,
            "price":81217.7,"direction":"buy"}]}}"#
                .to_string(),
        ))]);
        Htx::subscribe_trades(&mut htx, &symbol()).unwrap();

        let events = Htx::poll_events(&mut htx);
        let Event::Trade(trade) = &events[0] else {
            panic!("expected a trade, got {:?}", events[0]);
        };
        assert_eq!(trade.price, dec!(81217.7));
        assert_eq!(trade.quantity, dec!(2));
        assert_eq!(trade.aggressor, OrderSide::Buy);
    }

    /// The futures quote publishes its top of book as `[price, size]` pairs.
    ///
    /// Read as scalars they would fall back to zero, so a futures ticker would
    /// have reported a market with no bid and no ask -- a book that looks empty
    /// rather than one that was misread.
    #[test]
    fn a_futures_detail_frame_reads_its_paired_top_of_book() {
        let (mut htx, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"ch":"market.BTC-USDT.detail","ts":1788462493057,"tick":{"id":1788462480,
            "mrid":100127722487328,"open":81302.8,"close":81217.7,"high":81641.3,
            "low":80588,"amount":7221.06,"vol":7221060,"trade_turnover":571116798.5342,
            "count":67882,"ask":[81217.7,2514],"bid":[81217.6,767]}}"#
                .to_string(),
        ))]);
        Htx::subscribe_ticker(&mut htx, &symbol()).unwrap();

        let events = Htx::poll_events(&mut htx);
        let Event::Ticker(ticker) = &events[0] else {
            panic!("expected a ticker, got {:?}", events[0]);
        };
        assert_eq!(ticker.last, dec!(81217.7));
        assert_eq!(ticker.bid, dec!(81217.6));
        assert_eq!(ticker.ask, dec!(81217.7));
        assert_eq!(ticker.volume, dec!(7221.06));
    }

    /// The spot client keeps its own host, symbol form and quote channel.
    #[test]
    fn a_spot_client_still_subscribes_to_the_spot_stream() {
        let ws = Arc::new(MockWsTransport::new());
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut htx = Htx::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        Htx::subscribe_ticker(&mut htx, &symbol()).unwrap();

        assert_eq!(ws.connected_urls()[0], "wss://api.huobi.pro/ws");
        assert!(ws.sent()[0].contains("market.btcusdt.ticker"));
    }

    /// HTX closes a stream that does not answer its heartbeat.
    ///
    /// The public socket sends `{"ping": <ms>}` and the private v2 socket
    /// `{"action":"ping",...}`. Both were dropped as unrecognised -- they carry
    /// no `ch` -- so the venue closed the connection, the reconnect opened
    /// another, and that one was closed for the same reason. A stream that
    /// reconnects forever and delivers nothing.
    #[test]
    fn the_public_heartbeat_is_answered_with_the_same_value() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(r#"{"ping":1788462493057}"#.to_string()))]);
        Htx::subscribe_trades(&mut htx, &symbol()).unwrap();

        let events = Htx::poll_events(&mut htx);
        assert!(events.is_empty(), "a heartbeat is not an event: {events:?}");
        assert!(
            ws.sent().iter().any(|f| f == r#"{"pong":1788462493057}"#),
            "the heartbeat went unanswered: {:?}",
            ws.sent()
        );
    }

    #[test]
    fn the_private_heartbeat_is_answered_in_its_own_shape() {
        assert_eq!(
            pong_for(r#"{"action":"ping","data":{"ts":1788462493057}}"#).as_deref(),
            Some(r#"{"action":"pong","data":{"ts":1788462493057}}"#)
        );
        // An ordinary market frame is not a heartbeat.
        assert!(pong_for(r#"{"ch":"market.BTC-USDT.trade.detail","tick":{}}"#).is_none());
        assert!(pong_for("not json").is_none());
    }

    /// A real frame from the venue's own socket, in the shape it actually
    /// arrives -- recorded from `api.hbdm.com/linear-swap-ws`.
    #[test]
    fn a_real_trade_frame_parses() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"ch":"market.BTC-USDT.trade.detail","ts":1788462493057,"tick":{
            "id":100127722487328,"ts":1788462493055,"data":[{"amount":2,"quantity":0.002,
            "trade_turnover":162.4354,"ts":1788462493055,"id":1001277224873280000,
            "price":81217.7,"direction":"buy"}]}}"#
                .to_string(),
        ))]);
        Htx::subscribe_trades(&mut htx, &symbol()).unwrap();

        let events = Htx::poll_events(&mut htx);
        assert_eq!(events.len(), 1, "{events:?}");
        assert!(matches!(events[0], Event::Trade(_)));
    }

    #[test]
    fn subscribe_user_data_authenticates_and_streams_orders_and_accounts() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        ws.push_connection(vec![
            Ok(Some(
                r#"{"action":"req","code":200,"ch":"auth","data":{}}"#.to_string(),
            )),
            Ok(Some(
                r#"{"action":"sub","code":200,"ch":"orders#*"}"#.to_string(),
            )),
            Ok(Some(
                r#"{"action":"push","ch":"orders#btcusdt","data":{"orderId":55,"symbol":"btcusdt",
                "clientOrderId":"my","type":"buy-limit","orderStatus":"filled","orderSize":"1",
                "execAmt":"1","orderPrice":"100","tradePrice":"100"}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"action":"push","ch":"accounts.update#2","data":{"currency":"usdt",
                "balance":"950","available":"900"}}"#
                    .to_string(),
            )),
        ]);
        htx.subscribe_user_data().unwrap();
        assert_eq!(ws.connected_urls()[0], "wss://api.huobi.pro/ws/v2");
        assert!(ws.sent()[0].contains(r#""ch":"auth""#));
        assert!(ws.sent()[0].contains(r#""accessKey":"APIKEY""#));
        assert!(ws.sent()[0].contains(r#""signature""#));
        assert!(ws.sent()[1].contains(r#""ch":"orders#*""#));
        assert!(ws.sent()[2].contains(r#""ch":"accounts.update#2""#));

        let events = htx.poll_events();
        assert_eq!(events.len(), 2);
        let Event::OrderUpdate(order) = &events[0] else {
            panic!("first event must be an order update");
        };
        assert_eq!(order.id, "55");
        assert_eq!(order.client_order_id.as_deref(), Some("my"));
        assert_eq!(order.symbol, symbol());
        assert_eq!(order.side, OrderSide::Buy);
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.filled_quantity, dec!(1));
        assert_eq!(order.average_price, Some(dec!(100)));
        let Event::BalanceUpdate(balances) = &events[1] else {
            panic!("second event must be a balance update");
        };
        assert_eq!(balances.len(), 1);
        assert_eq!(balances[0].asset, "usdt");
        assert_eq!(balances[0].free, dec!(900));
        assert_eq!(balances[0].locked, dec!(50));
    }

    #[test]
    fn subscribe_user_data_requires_credentials() {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut htx = Htx::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        assert!(matches!(
            htx.subscribe_user_data().unwrap_err(),
            Error::InvalidCredentials(_)
        ));
    }

    #[test]
    fn keepalive_user_data_pings_the_private_stream() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        ws.push_connection(vec![]);
        htx.subscribe_user_data().unwrap();
        htx.keepalive_user_data().unwrap();
        assert!(ws.sent().iter().any(|f| f == r#"{"action":"ping"}"#));
    }

    #[test]
    fn keepalive_user_data_is_a_noop_before_subscribe() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        htx.keepalive_user_data().unwrap();
        assert_eq!(ws.sent(), Vec::<String>::new());
    }

    #[test]
    fn dropped_user_data_stream_reconnects_with_a_fresh_auth() {
        let (mut htx, ws) = signed_ws_client(1_700_000_000_000);
        // The first private connection closes on the first recv; the reconnect
        // target is a fresh open connection.
        ws.push_connection(vec![Ok(None)]);
        ws.push_connection(vec![]);
        htx.subscribe_user_data().unwrap();

        let events = htx.poll_events();
        assert!(events.contains(&Event::Disconnected));
        assert!(events.contains(&Event::Reconnected));
        // Two private connections (initial + reconnect), each re-signing the v2 auth.
        let auth_frames = ws
            .sent()
            .into_iter()
            .filter(|f| f.contains(r#""ch":"auth""#))
            .count();
        assert_eq!(auth_frames, 2);
        assert_eq!(ws.connected_urls().len(), 2);
        assert_eq!(ws.connected_urls()[1], "wss://api.huobi.pro/ws/v2");
    }

    #[test]
    fn ws_execution_is_a_documented_gap() {
        // HTX has no WebSocket order-entry API; the trait methods return a
        // documented error rather than faking a round trip.
        let (mut htx, _mock) = client();
        assert!(matches!(
            htx.place_order_ws(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
                .unwrap_err(),
            Error::Exchange { .. }
        ));
        assert!(matches!(
            htx.cancel_order_ws(&symbol(), "1").unwrap_err(),
            Error::Exchange { .. }
        ));
    }

    /// HTX stamps the envelope for the ticker and the `tick` for depth.
    /// Payloads captured from the live endpoints.
    #[test]
    fn the_public_reads_carry_the_venue_stamp() {
        let (htx, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"ch":"market.btcusdt.detail.merged","status":"ok","ts":1788396667304,"tick":{"close":77072.07,"bid":[77072.06,0.0009],"ask":[77072.07,0.019],"vol":95769370.0}}"#,
        );
        let ticker = htx.ticker(&symbol()).unwrap();
        assert_eq!(ticker.timestamp, 1_788_396_667_304);

        let (htx, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"ch":"market.btcusdt.depth.step0","status":"ok","ts":1788396667785,"tick":{"ts":1788396667002,"version":193172167185,"bids":[[77072.06,0.0009]],"asks":[[77072.07,0.019]]}}"#,
        );
        let book = htx.order_book(&symbol(), 5).unwrap();
        // The `tick`'s own stamp, not the envelope's: the envelope says when the
        // server replied, the tick says when the book was true.
        assert_eq!(book.timestamp, 1_788_396_667_002);
    }

    #[test]
    fn place_batch_reads_per_order_err_code() {
        let (htx, mock) = signed_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"id":42,"type":"spot","state":"working"}]}"#,
        );
        mock.push_json(
            200,
            r#"{"status":"ok","data":[
            {"order-id":1,"client-order-id":"","err-code":"","err-msg":""},
            {"order-id":0,"client-order-id":"","err-code":"account-frozen-balance-insufficient-error","err-msg":"no"}]}"#,
        );
        let results = htx
            .place_batch(&[
                OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)),
                OrderRequest::limit_buy(symbol(), dec!(1), dec!(101)),
            ])
            .unwrap();
        assert_eq!(results[0].as_ref().unwrap().id, "1");
        assert!(matches!(
            results[1].as_ref().unwrap_err(),
            Error::OrderRejected { .. }
        ));
        let reqs = mock.recorded_requests();
        assert!(reqs[1].url.contains("/v1/order/batch-orders"));
    }

    #[test]
    fn cancel_batch_is_one_call() {
        let (htx, mock) = signed_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"success":["1"],"failed":[]}}"#,
        );
        htx.cancel_batch(&symbol(), &["1".to_string(), "2".to_string()])
            .unwrap();
        let reqs = mock.recorded_requests();
        assert_eq!(reqs.len(), 1);
        assert!(reqs[0].url.contains("/v1/order/orders/batchcancel"));
        assert!(reqs[0]
            .body
            .as_ref()
            .unwrap()
            .contains(r#""order-ids":["1","2"]"#));
    }

    #[test]
    fn amend_and_oco_are_unsupported() {
        let (mut htx, _mock) = signed_client(0);
        assert!(matches!(
            AdvancedOrders::amend_order(&mut htx, &symbol(), "1", Some(dec!(1)), None).unwrap_err(),
            Error::Exchange { .. }
        ));
        assert!(matches!(
            AdvancedOrders::place_oco(
                &mut htx,
                &OcoRequest::new(symbol(), OrderSide::Sell, dec!(1), dec!(110), dec!(95))
            )
            .unwrap_err(),
            Error::Exchange { .. }
        ));
    }

    #[test]
    fn futures_ticker_targets_swap_host_and_contract_code() {
        let (htx, mock) = futures_client();
        mock.push_json(
            200,
            r#"{"status":"ok","tick":{"close":20000.5,"bid":[19999.0,1.0],"ask":[20001.0,2.0],"vol":1234.0}}"#,
        );
        let ticker = htx.ticker(&symbol()).unwrap();
        assert_eq!(ticker.last, dec!(20000.5));
        assert_eq!(ticker.bid, dec!(19999));
        assert_eq!(
            mock.recorded_requests()[0].url,
            "https://api.hbdm.com/linear-swap-ex/market/detail/merged?contract_code=BTC-USDT"
        );
    }

    #[test]
    fn futures_market_order_uses_cross_swap_family() {
        let (htx, mock) = signed_futures_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"order_id":123,"order_id_str":"123"}}"#,
        );
        let order = htx
            .place_order(&OrderRequest::market_buy(symbol(), dec!(2)))
            .unwrap();
        assert_eq!(order.id, "123");
        assert_eq!(order.status, OrderStatus::New);
        let req = &mock.recorded_requests()[0];
        assert!(req
            .url
            .contains("api.hbdm.com/linear-swap-api/v1/swap_cross_order"));
        let body = req.body.as_ref().unwrap();
        assert!(body.contains(r#""contract_code":"BTC-USDT""#));
        assert!(body.contains(r#""direction":"buy""#));
        assert!(body.contains(r#""offset":"open""#));
        assert!(body.contains(r#""order_price_type":"optimal_5""#));
        assert!(body.contains(r#""volume":2"#));
        assert!(body.contains(r#""lever_rate":1"#));
    }

    #[test]
    fn futures_query_order_uses_swap_order_info() {
        let (htx, mock) = signed_futures_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"order_id":88,"order_id_str":"88","direction":"buy",
            "offset":"open","order_price_type":"limit","status":6,"volume":3,"trade_volume":3,
            "price":100,"trade_avg_price":100}]}"#,
        );
        let order = htx.query_order(&symbol(), "88").unwrap();
        assert_eq!(order.id, "88");
        assert_eq!(order.side, OrderSide::Buy);
        assert_eq!(order.order_type, OrderType::Limit);
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.quantity, dec!(3));
        let req = &mock.recorded_requests()[0];
        assert!(req
            .url
            .contains("/linear-swap-api/v1/swap_cross_order_info"));
        assert!(req
            .body
            .as_ref()
            .unwrap()
            .contains(r#""contract_code":"BTC-USDT""#));
    }

    #[test]
    fn futures_cancel_and_open_orders_use_swap_endpoints() {
        let (htx, mock) = signed_futures_client(0);
        mock.push_json(200, r#"{"status":"ok","data":{"successes":"88"}}"#);
        htx.cancel_order(&symbol(), "88").unwrap();
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"orders":[{"order_id":90,"order_id_str":"90",
            "direction":"sell","offset":"open","order_price_type":"limit","status":3,
            "volume":5,"trade_volume":2,"price":21000,"trade_avg_price":21000}]}}"#,
        );
        let orders = htx.open_orders(Some(&symbol())).unwrap();
        assert_eq!(orders.len(), 1);
        assert_eq!(orders[0].side, OrderSide::Sell);
        assert_eq!(orders[0].status, OrderStatus::PartiallyFilled);
        let reqs = mock.recorded_requests();
        assert!(reqs[0]
            .url
            .contains("/linear-swap-api/v1/swap_cross_cancel"));
        assert!(reqs[1]
            .url
            .contains("/linear-swap-api/v1/swap_cross_openorders"));
    }

    #[test]
    fn derivatives_positions_parse_cross() {
        let (mut htx, mock) = signed_futures_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"contract_code":"BTC-USDT","direction":"buy","volume":3,
            "cost_open":20000,"last_price":20100,"lever_rate":10,"profit_unreal":30}]}"#,
        );
        let positions = Derivatives::positions(&mut htx, Some(&symbol())).unwrap();
        assert_eq!(positions.len(), 1);
        assert_eq!(positions[0].symbol, Symbol::new("BTC", "USDT"));
        assert_eq!(positions[0].side, PositionSide::Long);
        assert_eq!(positions[0].quantity, dec!(3));
        assert_eq!(positions[0].entry_price, dec!(20000));
        assert_eq!(positions[0].mark_price, dec!(20100));
        assert_eq!(positions[0].leverage, dec!(10));
        assert_eq!(positions[0].margin_mode, MarginMode::Cross);
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/linear-swap-api/v1/swap_cross_position_info"));
    }

    #[test]
    fn set_leverage_persists_and_flows_into_orders() {
        let (htx, mock) = signed_futures_client(0);
        mock.push_json(200, r#"{"status":"ok"}"#);
        htx.set_leverage(&symbol(), 5).unwrap();
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/linear-swap-api/v1/swap_cross_switch_lever_rate"));

        mock.push_json(
            200,
            r#"{"status":"ok","data":{"order_id":1,"order_id_str":"1"}}"#,
        );
        htx.place_order(&OrderRequest::market_buy(symbol(), dec!(1)))
            .unwrap();
        let reqs = mock.recorded_requests();
        assert!(reqs[1].body.as_ref().unwrap().contains(r#""lever_rate":5"#));
    }

    #[test]
    fn set_margin_mode_isolated_is_unsupported() {
        let (htx, _mock) = signed_futures_client(0);
        assert!(htx.set_margin_mode(&symbol(), MarginMode::Cross).is_ok());
        assert!(matches!(
            htx.set_margin_mode(&symbol(), MarginMode::Isolated)
                .unwrap_err(),
            Error::Exchange { .. }
        ));
    }

    /// The swap `order_price_type` carries the time-in-force and post-only in one
    /// field, and the futures market order is `optimal_5` -- IOC-shaped, with no
    /// fill-or-kill spelling.
    #[test]
    fn futures_orders_carry_the_time_in_force_or_refuse_it() {
        let (htx, mock) = signed_futures_client(1000);
        mock.push_json(200, "{}");
        let ioc = OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
            .with_time_in_force(TimeInForce::Ioc);
        let _ = htx.place_order(&ioc);
        let body = mock.recorded_requests()[0].body.clone().unwrap_or_default();
        assert!(body.contains(r#""order_price_type":"ioc""#));

        let (htx, mock) = signed_futures_client(1000);
        mock.push_json(200, "{}");
        let fok = OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
            .with_time_in_force(TimeInForce::Fok);
        let _ = htx.place_order(&fok);
        let body = mock.recorded_requests()[0].body.clone().unwrap_or_default();
        assert!(body.contains(r#""order_price_type":"fok""#));

        let (htx, mock) = signed_futures_client(1000);
        mock.push_json(200, "{}");
        let poc = OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)).post_only();
        let _ = htx.place_order(&poc);
        let body = mock.recorded_requests()[0].body.clone().unwrap_or_default();
        assert!(body.contains(r#""order_price_type":"post_only""#));

        // Post-only and a non-GTC time-in-force are both values of that one
        // field, so asking for both is refused rather than half-honoured.
        let (htx, mock) = signed_futures_client(1000);
        let both = OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
            .post_only()
            .with_time_in_force(TimeInForce::Ioc);
        let err = htx.place_order(&both).unwrap_err();
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert_eq!(mock.recorded_requests(), Vec::new());

        let (htx, mock) = signed_futures_client(1000);
        let market_fok =
            OrderRequest::market_buy(symbol(), dec!(1)).with_time_in_force(TimeInForce::Fok);
        let err = htx.place_order(&market_fok).unwrap_err();
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert_eq!(mock.recorded_requests(), Vec::new());
    }

    /// Post-only is the `limit-maker` order kind on spot, so it cannot ride on a
    /// market order.
    #[test]
    fn spot_refuses_post_only_on_a_market_order() {
        let (htx, mock) = signed_client(1000);
        let request = OrderRequest::market_buy(symbol(), dec!(1)).post_only();
        let err = htx.place_order(&request).unwrap_err();
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert_eq!(mock.recorded_requests(), Vec::new());
    }

    #[test]
    fn close_position_is_reduce_only_opposite() {
        let (mut htx, mock) = signed_futures_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"contract_code":"BTC-USDT","direction":"buy","volume":3,
            "cost_open":20000,"last_price":20100,"lever_rate":10,"profit_unreal":30}]}"#,
        );
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"order_id":9,"order_id_str":"9"}}"#,
        );
        Derivatives::close_position(&mut htx, &symbol()).unwrap();
        let reqs = mock.recorded_requests();
        let body = reqs[1].body.as_ref().unwrap();
        assert!(body.contains(r#""direction":"sell""#));
        assert!(body.contains(r#""offset":"close""#));
    }

    #[test]
    fn futures_balances_split_available_and_frozen() {
        let (htx, mock) = signed_futures_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"margin_asset":"USDT","margin_available":800,"margin_frozen":200}]}"#,
        );
        let bals = htx.balances().unwrap();
        assert_eq!(bals[0].asset, "USDT");
        assert_eq!(bals[0].free, dec!(800));
        assert_eq!(bals[0].locked, dec!(200));
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/linear-swap-api/v1/swap_cross_account_info"));
    }

    #[test]
    fn iso_and_encode() {
        assert_eq!(iso8601_no_millis(0), "1970-01-01T00:00:00");
        assert_eq!(iso8601_no_millis(1_700_000_000_000), "2023-11-14T22:13:20");
        assert_eq!(encode("2023-11-14T22:13:20"), "2023-11-14T22%3A13%3A20");
        assert_eq!(encode("a+b/c=d"), "a%2Bb%2Fc%3Dd");
    }

    #[test]
    fn ticker_reads_numeric_json() {
        let (htx, mock) = client();
        mock.push_json(
            200,
            r#"{"status":"ok","tick":{"close":20000.5,"bid":[19999.0,1.0],"ask":[20001.0,2.0],"vol":1234.0}}"#,
        );
        let ticker = htx.ticker(&symbol()).unwrap();
        assert_eq!(ticker.last, dec!(20000.5));
        assert_eq!(ticker.bid, dec!(19999));
        assert_eq!(ticker.ask, dec!(20001));
        assert_eq!(
            mock.recorded_requests()[0].url,
            "https://api.huobi.pro/market/detail/merged?symbol=btcusdt"
        );
    }

    #[test]
    fn klines_reversed_named_fields() {
        let (htx, mock) = client();
        mock.push_json(
            200,
            r#"{"status":"ok","data":[
            {"id":1700000060,"open":105,"close":105.5,"low":104,"high":106,"amount":2},
            {"id":1700000000,"open":100,"close":105,"low":95,"high":110,"amount":12}]}"#,
        );
        let candles = htx.klines(&symbol(), "1h", 2).unwrap();
        assert_eq!(candles[0].timestamp, 1_700_000_000);
        assert!((candles[0].high - 110.0).abs() < 1e-9);
        assert!((candles[0].low - 95.0).abs() < 1e-9);
    }

    #[test]
    fn order_book_numeric_levels() {
        let (htx, mock) = client();
        mock.push_json(
            200,
            r#"{"status":"ok","tick":{"version":77,"bids":[[100.0,1.5]],"asks":[[101.0,2.0]]}}"#,
        );
        let book = htx.order_book(&symbol(), 5).unwrap();
        assert_eq!(book.last_update_id, 77);
        assert_eq!(book.bids[0], BookLevel::new(dec!(100), dec!(1.5)));
    }

    #[test]
    fn error_status_maps() {
        let (htx, mock) = client();
        mock.push_json(
            200,
            r#"{"status":"error","err-code":"account-frozen-balance-insufficient-error","err-msg":"no"}"#,
        );
        assert!(matches!(
            htx.ticker(&symbol()).unwrap_err(),
            Error::InsufficientBalance
        ));
    }

    #[test]
    fn place_order_fetches_account_and_signs() {
        let (htx, mock) = signed_client(0);
        // First the account lookup, then the order.
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"id":42,"type":"spot","state":"working"}]}"#,
        );
        mock.push_json(200, r#"{"status":"ok","data":"777"}"#);
        let order = htx
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        assert_eq!(order.id, "777");
        assert_eq!(order.status, OrderStatus::New);

        let place = &mock.recorded_requests()[1];
        assert_eq!(place.method, HttpMethod::Post);
        // Reconstruct the AWS-style signature over the sorted auth params.
        let ts = "1970-01-01T00:00:00";
        let encoded = format!(
            "AccessKeyId=APIKEY&SignatureMethod=HmacSHA256&SignatureVersion=2&Timestamp={}",
            encode(ts)
        );
        let canonical = format!("POST\napi.huobi.pro\n/v1/order/orders/place\n{encoded}");
        let sign = hmac_sha256_base64(b"SECRET", canonical.as_bytes());
        assert!(place.url.contains(&format!("Signature={}", encode(&sign))));
        // The account-id was cached (order body carries it).
        assert!(place
            .body
            .as_ref()
            .unwrap()
            .contains(r#""account-id":"42""#));
    }

    #[test]
    fn query_order_parses_type_and_state() {
        let (htx, mock) = signed_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"id":777,"client-order-id":"","symbol":"btcusdt",
            "type":"sell-limit","state":"filled","amount":"2","field-amount":"2","price":"100"}}"#,
        );
        let order = htx.query_order(&symbol(), "777").unwrap();
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.side, OrderSide::Sell);
        assert_eq!(order.order_type, OrderType::Limit);
        assert_eq!(order.filled_quantity, dec!(2));
    }

    #[test]
    fn balances_merge_trade_and_frozen() {
        let (htx, mock) = signed_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"id":42,"type":"spot","state":"working"}]}"#,
        );
        mock.push_json(
            200,
            r#"{"status":"ok","data":{"list":[
            {"currency":"usdt","type":"trade","balance":"100.5"},
            {"currency":"usdt","type":"frozen","balance":"25.5"}]}}"#,
        );
        let bals = htx.balances().unwrap();
        assert_eq!(bals.len(), 1);
        assert_eq!(bals[0].asset, "usdt");
        assert_eq!(bals[0].free, dec!(100.5));
        assert_eq!(bals[0].locked, dec!(25.5));
    }

    #[test]
    fn signed_requires_credentials() {
        let (htx, _) = client();
        assert!(matches!(
            htx.query_order(&symbol(), "1").unwrap_err(),
            Error::InvalidCredentials(_)
        ));
    }

    #[test]
    fn ws_parses_trade_and_depth() {
        let ws = Arc::new(MockWsTransport::new());
        ws.push_connection(vec![
            Ok(Some(
                r#"{"ch":"market.btcusdt.trade.detail","tick":{"data":[
                {"price":100.0,"amount":0.5,"direction":"buy","ts":1}]}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"ch":"market.btcusdt.depth.step0","tick":{"version":9,
                "bids":[[100.0,1.0]],"asks":[[101.0,2.0]]}}"#
                    .to_string(),
            )),
            Ok(Some(r#"{"ping":123}"#.to_string())),
        ]);
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut htx = Htx::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        htx.subscribe_trades(&symbol()).unwrap();
        assert!(ws.sent()[0].contains(r#""sub":"market.btcusdt.trade.detail""#));

        let events = htx.poll_events();
        assert_eq!(events.len(), 2);
        let Event::Trade(t) = &events[0] else {
            panic!("expected trade")
        };
        assert_eq!(t.aggressor, OrderSide::Buy);
        assert!(matches!(events[1], Event::BookDelta(_)));
    }

    #[test]
    fn works_as_a_boxed_exchange() {
        let (htx, mock) = signed_client(0);
        mock.push_json(
            200,
            r#"{"status":"ok","data":[{"id":42,"type":"spot","state":"working"}]}"#,
        );
        mock.push_json(200, r#"{"status":"ok","data":"1"}"#);
        let mut exchange: Box<dyn Exchange> = Box::new(htx);
        assert_eq!(exchange.name(), "htx");
        let order = exchange
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        assert_eq!(order.id, "1");
    }

    #[test]
    fn system_clock_is_sane() {
        assert!(system_now_ms() > 1_600_000_000_000);
    }

    /// `Debug` reports connection state, never secret material. A client is
    /// formatted into logs and error messages, so anything it prints is
    /// somewhere a credential must not be.
    #[test]
    fn debug_reports_state_without_credentials() {
        let (client, _http) = signed_client(1_700_000_000_000);
        let rendered = format!("{client:?}");

        assert!(rendered.starts_with("Htx {"));
        assert!(rendered.contains("authenticated: true"));
        assert!(!rendered.contains("SECRET"));
        assert!(!rendered.contains("APIKEY"));
    }

    /// The clock offset must reach the wire, not just the struct.
    ///
    /// A venue refuses a signed request whose timestamp is outside its own
    /// receive window, so a machine a few seconds off has every order rejected
    /// -- with a message about the window, not about the clock. This asserts the
    /// whole path: sync, then a signed request carrying the adjusted time.
    #[test]
    fn sync_time_shifts_signed_timestamps() {
        let (mut htx, http) = signed_client(1_000_000);

        // Shape verified against the live public endpoint on 2026-09-01.
        http.push_json(200, r#"{"status":"ok","data":1004500}"#);
        let offset = htx.sync_time().expect("sync must succeed");
        assert_eq!(offset, 4_500, "the venue is 4.5 s ahead of this machine");

        http.push_json(200, r#"{"status":"ok","data":[]}"#);
        let _ = htx.balances();

        let requests = http.recorded_requests();
        let signed = requests.last().expect("a signed request was recorded");
        // HTX signs an ISO-8601 timestamp in the query; 1_004_500 ms is
        // 1970-01-01T00:16:44.
        assert!(
            signed.url.contains("00%3A16%3A44") || signed.url.contains("00:16:44"),
            "the signed query must carry the venue's time: {}",
            signed.url
        );
    }

    /// A rate-limited response carries the venue's advised wait.
    ///
    /// The limit itself is recognised from the body -- a code in the error
    /// envelope -- while the wait arrives in the `Retry-After` header, so the
    /// two are read in different places. Until they were joined, every
    /// `RateLimited` this client raised carried `retry_after: None`: a field
    /// the error type documents, and that nothing in the crate ever filled.
    #[test]
    fn rate_limit_carries_the_venues_advised_wait() {
        let (htx, mock) = client();
        mock.push_response(
            crate::transport::HttpResponse::new(
                200,
                r#"{"status":"error","err-code":"rate-limit","err-msg":"slow"}"#,
            )
            .with_header("Retry-After", "2.5"),
        );
        let err = htx.ticker(&symbol()).unwrap_err();
        let wait = std::time::Duration::from_millis(2500);
        assert!(matches!(err, Error::RateLimited { retry_after: Some(d) } if d == wait));
    }
}