wickra-exchange-core 0.2.1

//! KuCoin (v1/v2 API) — the fifth exchange.
//!
//! KuCoin signs with base64(HMAC-SHA256) over `msTimestamp + METHOD + endpoint +
//! body` under `KC-API-*` headers, and — unlike OKX/Bitget — the passphrase
//! itself is HMAC-signed (`KC-API-PASSPHRASE = base64(HMAC-SHA256(secret,
//! passphrase))`, key version 2). Symbols are dash-form (`BTC-USDT`); the
//! envelope success code is the string `"200000"`. KuCoin candles are
//! newest-first and ordered `[time, open, close, high, low, volume, turnover]`.
//!
//! [`AdvancedOrders`]: STP via the `stp` flag on order create, native spot batch
//! place (`/api/v1/orders/multi`, per-order results), native OCO
//! (`/api/v3/oco/order`, returned as one order-list). KuCoin has no
//! batch-cancel-by-id (so `cancel_batch` cancels sequentially) and no in-place
//! amend (`amend_order` is a documented gap).

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, LongShortRatio, MarkIndex, OpenInterest,
};
use crate::idempotency::ClientIdGenerator;
use crate::normalize::{format_decimal, parse_decimal};
use crate::options::{ExchangeOptions, MarginMode, MarketType, PositionMode, 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::FromPrimitive;
use rust_decimal::Decimal;
use serde::Deserialize;
use std::collections::HashMap;
use std::fmt;
use wickra_core::Candle;

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
}

/// A KuCoin client over injected transports.
pub struct KuCoin {
    http: Box<dyn HttpTransport>,
    ws: Option<Box<dyn WsTransport>>,
    rest_base: String,
    market_type: MarketType,
    /// One-way or hedge. KuCoin Futures is one-way only, so a hedged
    /// configuration is refused rather than silently ignored.
    position_mode: PositionMode,
    credentials: Option<Credentials>,
    now_ms: Box<dyn Fn() -> i64 + Send + Sync>,
    /// Client order ids for orders the caller did not name.
    ///
    /// Seeded from the clock at construction and monotonic from there: the
    /// seed keeps two clients in one process apart, the counter keeps two
    /// orders in one millisecond apart. The previous fallback was the
    /// millisecond alone, so a second order placed inside the same
    /// millisecond carried an id the venue had already seen and was refused
    /// as a duplicate.
    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)>,
    /// Which derivatives channels are subscribed, per contract.
    ///
    /// `/contract/instrument` carries both subjects, so one subscription
    /// delivers frames for the other; only what was asked for is emitted.
    derivatives_channels: Vec<(String, DerivativesChannel)>,
    /// Leverage applied to futures orders. KuCoin sets leverage per order rather
    /// than per account, so [`set_leverage`](Self::set_leverage) records it here.
    leverage: 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>>,
    /// Set once the private stream is subscribed, so [`poll_events`](Self::poll_events)
    /// re-subscribes it (re-negotiating a fresh bullet token) 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 KuCoin {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("KuCoin")
            .field("ws", &self.ws.is_some())
            .field("rest_base", &self.rest_base)
            .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("leverage", &self.leverage)
            .field("private_connection", &self.private_connection.is_some())
            .field("user_data_active", &self.user_data_active)
            .finish_non_exhaustive()
    }
}

impl KuCoin {
    /// 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("KuCoin", self.market_type, super::SPOT_AND_LINEAR)
    }

    fn build(
        http: Box<dyn HttpTransport>,
        options: &ExchangeOptions,
        credentials: Option<Credentials>,
    ) -> Self {
        let futures = options.market_type.is_derivatives();
        Self {
            http,
            ws: None,
            rest_base: if futures {
                "https://api-futures.kucoin.com".to_string()
            } else {
                "https://api.kucoin.com".to_string()
            },
            market_type: options.market_type,
            position_mode: options.position_mode,
            credentials,
            now_ms: Box::new(system_now_ms),
            client_ids: ClientIdGenerator::with_seed(
                "wkex",
                u64::try_from(system_now_ms()).unwrap_or(0),
            ),
            clock: ServerClock::new(),
            connection: None,
            sub_messages: Vec::new(),
            subscriptions: Vec::new(),
            derivatives_channels: Vec::new(),
            leverage: 1,
            private_connection: None,
            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 /api/v1/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> {
        // unwrap_envelope hands back `data`, which here is a bare number rather
        // than an object: {"code":"200000","data":1788295957789}.
        let local_before = (self.now_ms)();
        let value = self.get("/api/v1/timestamp", "")?;
        let server_ms = value
            .as_i64()
            .ok_or_else(|| Error::Deserialization("timestamp: data not an integer".into()))?;
        self.clock.sync(local_before, server_ms);
        Ok(self.clock.offset_ms())
    }

    /// Whether this client targets KuCoin Futures (a separate host and API with
    /// contract symbols like `XBTUSDTM`) rather than spot.
    fn is_futures(&self) -> bool {
        self.market_type.is_derivatives()
    }

    /// Refuse a futures order on a client configured for hedge mode.
    ///
    /// KuCoin Futures holds one net position per contract and has no
    /// hedge (dual-side) mode, so there is no
    /// field on the order that could carry the side. Placing the order anyway
    /// would open or close the single net position -- which is what the caller
    /// asked to avoid by configuring hedge. Same treatment as
    /// `set_margin_mode(Isolated)` here: say so rather than do something else.
    ///
    /// # Errors
    /// Returns [`Error::Exchange`] on a futures client in
    /// [`PositionMode::Hedge`](crate::PositionMode).
    fn ensure_one_way(&self) -> Result<()> {
        if self.is_futures() && self.position_mode == PositionMode::Hedge {
            return Err(Error::Exchange {
                code: "unsupported".to_string(),
                message: "KuCoin Futures is one-way only; PositionMode::Hedge cannot be                           expressed on its orders"
                    .to_string(),
            });
        }
        Ok(())
    }

    /// The KuCoin **futures** contract symbol for a canonical [`Symbol`]:
    /// `BTC/USDT` -> `XBTUSDTM` (KuCoin uses `XBT` for Bitcoin and a trailing
    /// `M` for the USDⓈ-M perpetual).
    fn futures_symbol(symbol: &Symbol) -> String {
        let base = if symbol.base() == "BTC" {
            "XBT"
        } else {
            symbol.base()
        };
        format!("{base}{}M", symbol.quote())
    }

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

    /// Build an authenticated KuCoin client (credentials must carry a passphrase).
    #[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 KuCoin wire symbol for a canonical [`Symbol`] (`BTC/USDT` -> `BTC-USDT`).
    #[must_use]
    pub fn wire_symbol(symbol: &Symbol) -> String {
        format!("{}-{}", symbol.base(), symbol.quote())
    }

    /// A ticker for `symbol` (from the 24-hour stats endpoint).
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the symbol is unknown.
    pub fn ticker(&self, symbol: &Symbol) -> Result<Ticker> {
        let query = format!("symbol={}", Self::wire_symbol(symbol));
        let data = self.get("/api/v1/market/stats", &query)?;
        let raw: RawStats = parse_json(data)?;
        Ok(Ticker {
            symbol: symbol.clone(),
            last: parse_decimal(&raw.last)?,
            bid: parse_decimal(&raw.buy)?,
            ask: parse_decimal(&raw.sell)?,
            volume: parse_decimal(&raw.vol)?,
            timestamp: raw.time,
        })
    }

    /// Up to `limit` candles for `symbol` at `interval` (unified). KuCoin 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 query = format!(
            "symbol={}&type={}",
            Self::wire_symbol(symbol),
            map_type(interval),
        );
        let data = self.get("/api/v1/market/candles", &query)?;
        let rows: Vec<Vec<String>> = parse_json(data)?;
        let mut candles = rows
            .iter()
            .map(|row| parse_kline_row(row))
            .collect::<Result<Vec<_>>>()?;
        candles.reverse();
        Ok(candles)
    }

    /// A depth snapshot of `symbol` (top 20 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 query = format!("symbol={}", Self::wire_symbol(symbol));
        let data = self.get("/api/v1/market/orderbook/level2_20", &query)?;
        let raw: RawDepth = parse_json(data)?;
        Ok(OrderBookSnapshot {
            symbol: symbol.clone(),
            last_update_id: raw.sequence.parse().unwrap_or(0),
            bids: parse_levels(&raw.bids)?,
            asks: parse_levels(&raw.asks)?,
            timestamp: raw.time,
        })
    }

    /// Subscribe to a pushed derivatives channel.
    ///
    /// KuCoin publishes both on `/contract/instrument`, under two subjects:
    /// `funding.rate` and `mark.index.price`. The funding subject carries no
    /// price, so a funding print reports a mark of zero rather than one taken
    /// from the other subject at another moment.
    ///
    /// `Liquidations` is refused: KuCoin publishes no public stream of forced
    /// orders.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, for `Liquidations`, or if the
    /// subscription fails.
    pub fn subscribe_derivatives(
        &mut self,
        symbol: &Symbol,
        channel: DerivativesChannel,
    ) -> Result<()> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "KuCoin",
                "a derivatives channel on a spot client",
                "funding and mark/index exist only on KuCoin Futures",
            ));
        }
        if channel == DerivativesChannel::Liquidations {
            return Err(Error::unsupported_field(
                "KuCoin",
                "a liquidations channel",
                "KuCoin publishes no public stream of forced orders",
            ));
        }
        let contract = Self::futures_symbol(symbol);
        if !self
            .derivatives_channels
            .iter()
            .any(|(c, k)| c == &contract && *k == channel)
        {
            self.derivatives_channels.push((contract, channel));
        }
        self.subscribe_instrument(symbol)
    }

    /// Subscribe to `/contract/instrument`, which is not a `/market/` topic and
    /// so does not go through the spot-topic mapping.
    fn subscribe_instrument(&mut self, symbol: &Symbol) -> Result<()> {
        self.ensure_market_is_routed()?;
        let wire = Self::futures_symbol(symbol);
        if self.connection.is_none() {
            let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
            let connection = ws.connect(FUTURES_WS_URL)?;
            self.connection = Some(connection);
        }
        let id = (self.now_ms)();
        let message = format!(
            r#"{{"id":"{id}","type":"subscribe","topic":"/contract/instrument:{wire}","response":true}}"#
        );
        self.connection
            .as_mut()
            .expect("connection just ensured")
            .send(&message)?;
        if !self.sub_messages.contains(&message) {
            self.sub_messages.push(message);
        }
        if !self.subscriptions.iter().any(|(w, _)| w == &wire) {
            self.subscriptions.push((wire, symbol.clone()));
        }
        Ok(())
    }

    /// The current open interest (`GET /api/v1/contracts/<contract>`).
    ///
    /// Polled, not pushed. KuCoin reports it in contracts and publishes the
    /// contract's `multiplier` in the same reply, so the base-currency figure
    /// is exact arithmetic on one reading rather than two.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, if the request fails, or if the
    /// reply carries neither figure.
    pub fn open_interest(&self, symbol: &Symbol) -> Result<OpenInterest> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "KuCoin",
                "open interest on a spot client",
                "open interest is a futures figure",
            ));
        }
        let path = format!("/api/v1/contracts/{}", Self::futures_symbol(symbol));
        let data = self.get(&path, "")?;
        let contracts = parse_decimal(field_str(&data, "openInterest")?)?;
        let multiplier = data
            .get("multiplier")
            .and_then(serde_json::Value::as_f64)
            .and_then(|m| parse_decimal(&m.to_string()).ok())
            .ok_or_else(|| Error::Deserialization("no contract multiplier".to_string()))?;
        Ok(OpenInterest {
            symbol: symbol.clone(),
            open_interest: contracts * multiplier,
            timestamp: 0,
        })
    }

    /// KuCoin publishes no public long/short positioning figure.
    ///
    /// # Errors
    /// Always returns an [`Error`]: there is nothing to read.
    pub fn long_short_ratio(&self, _symbol: &Symbol) -> Result<LongShortRatio> {
        Err(Error::unsupported_field(
            "KuCoin",
            "long/short positioning",
            "KuCoin publishes no public account-positioning figure",
        ))
    }

    /// 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<()> {
        self.subscribe(symbol, "/market/match")
    }

    /// Subscribe to the order-book stream for `symbol`.
    ///
    /// # Errors
    /// See [`subscribe_trades`](Self::subscribe_trades).
    pub fn subscribe_book(&mut self, symbol: &Symbol) -> Result<()> {
        self.subscribe(symbol, "/market/level2")
    }

    /// Subscribe to the ticker stream for `symbol`.
    ///
    /// # Errors
    /// See [`subscribe_trades`](Self::subscribe_trades).
    pub fn subscribe_ticker(&mut self, symbol: &Symbol) -> Result<()> {
        self.subscribe(symbol, "/market/ticker")
    }

    fn subscribe(&mut self, symbol: &Symbol, topic_prefix: &str) -> Result<()> {
        self.ensure_market_is_routed()?;
        // Spot and futures are different hosts with differently named topics.
        // The topic arrives spelled `/market/<name>`; on a futures client it is
        // mapped once here rather than at each call site.
        let futures = self.is_futures();
        let (topic_prefix, wire) = if futures {
            (futures_topic(topic_prefix), Self::futures_symbol(symbol))
        } else {
            (topic_prefix, Self::wire_symbol(symbol))
        };
        if self.connection.is_none() {
            // The bullet-token negotiation and instance endpoint are handled by
            // the real transport adapter; the module only produces the topics.
            let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
            let url = if futures {
                FUTURES_WS_URL
            } else {
                "wss://ws-api-spot.kucoin.com/"
            };
            let connection = ws.connect(url)?;
            self.connection = Some(connection);
        }
        let id = (self.now_ms)();
        let message = format!(
            r#"{{"id":"{id}","type":"subscribe","topic":"{topic_prefix}:{wire}","response":true}}"#
        );
        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.
    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(|| wire.parse().unwrap_or_else(|_| Symbol::new(wire, "")))
        };
        let channels = self.derivatives_channels.clone();
        let subscribed = |wire: &str, channel: DerivativesChannel| {
            channels.iter().any(|(c, k)| c == wire && *k == channel)
        };
        let mut events = Vec::new();
        if let Some(connection) = self.connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Ok(Some(event)) = parse_ws_message(&frame, &resolve, &subscribed) {
                    events.push(event);
                }
            }
        }
        // Drain the private user-data stream (tradeOrders/account.balance), if open.
        if let Some(connection) = self.private_connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Ok(Some(event)) = parse_ws_message(&frame, &resolve, &subscribed) {
                    events.push(event);
                }
            }
        }
        // A dropped private stream is re-subscribed by re-negotiating a fresh
        // bullet-private token over REST (the previous token is single-use and
        // time-bound, so it cannot be replayed).
        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://ws-api-spot.kucoin.com/";
        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. Negotiates a bullet-private token over
    /// REST (`POST /api/v1/bullet-private`, signed), connects the returned
    /// instance endpoint (`<endpoint>?token=<token>&connectId=<id>`), then
    /// subscribes to the account's own order and balance channels -- on spot
    /// `/spotMarket/tradeOrders` and `/account/balance`, on futures
    /// `/contractMarket/tradeOrders` and `/contractAccount/wallet`. A futures
    /// client negotiates its token against the futures host, which serves the
    /// same `/api/v1/bullet-private` path. Afterwards [`poll_events`](Self::poll_events) also surfaces the
    /// account's own [`Event::OrderUpdate`] and [`Event::BalanceUpdate`].
    ///
    /// The bullet token expires;
    /// [`keepalive_user_data`](Self::keepalive_user_data) sends KuCoin's ping to
    /// keep it alive, and a dropped stream is re-subscribed automatically on the
    /// next [`poll_events`](Self::poll_events), which re-negotiates a fresh token.
    ///
    /// # Errors
    /// Returns [`Error::InvalidCredentials`] without credentials or a passphrase,
    /// [`Error::NotConnected`] without a WebSocket transport, or another
    /// [`Error`] if the token negotiation or subscription 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. `rest_base` already points at
        // the market's host, and both serve the same negotiation path.
        let (orders_topic, account_topic) = if self.is_futures() {
            ("/contractMarket/tradeOrders", "/contractAccount/wallet")
        } else {
            ("/spotMarket/tradeOrders", "/account/balance")
        };
        let data = self.signed_request(HttpMethod::Post, "/api/v1/bullet-private", "", "")?;
        let bullet: BulletToken = parse_json(data)?;
        let server = bullet.instance_servers.into_iter().next().ok_or_else(|| {
            Error::NotFound("bullet-private returned no instance server".to_string())
        })?;
        let connect_id = (self.now_ms)();
        let url = format!(
            "{}?token={}&connectId={connect_id}",
            server.endpoint, bullet.token
        );
        let orders = format!(
            r#"{{"id":"{connect_id}","type":"subscribe","topic":"{orders_topic}","privateChannel":true,"response":true}}"#
        );
        let account = format!(
            r#"{{"id":"{}","type":"subscribe","topic":"{account_topic}","privateChannel":true,"response":true}}"#,
            connect_id + 1
        );
        let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
        let mut connection = ws.connect(&url)?;
        connection.send(&orders)?;
        connection.send(&account)?;
        self.private_connection = Some(connection);
        self.user_data_active = true;
        Ok(())
    }

    /// Send KuCoin's application-level heartbeat (`{"id":..,"type":"ping"}`) on the
    /// private stream so the bullet token 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<()> {
        let ping = format!(r#"{{"id":"{}","type":"ping"}}"#, (self.now_ms)());
        if let Some(connection) = self.private_connection.as_mut() {
            connection.send(&ping)?;
        }
        Ok(())
    }

    /// Place an order. KuCoin requires a client order id; one is generated from
    /// the clock when the request does not supply it.
    ///
    /// # 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(
                "KuCoin",
                "reduce_only on a spot order",
                "spot holds balances, not positions, and has none to reduce",
            ));
        }
        Ok(())
    }

    /// Add KuCoin's trigger fields to an order body.
    ///
    /// `stop` says **which way the market must cross the trigger**, and KuCoin
    /// will not infer it. `loss` fires when the market moves against the side
    /// -- down for a sell, up for a buy -- which is what a stop-loss means;
    /// `entry` fires the other way, which is a breakout entry. The wrong one is
    /// a different order that happens to use the same price.
    ///
    /// On futures, `stopPriceType` names the price that arms it: `MP` is the
    /// mark, which is what a position is liquidated against, so it is what a
    /// stop there is meant to watch. Spot has no mark and takes no such field.
    fn add_trigger(&self, body: &mut serde_json::Value, request: &OrderRequest) -> Result<()> {
        let stop = request
            .stop_price
            .ok_or(Error::InvalidOrder("a trigger order requires a stop price"))?;
        body["stop"] = serde_json::json!(match request.side {
            OrderSide::Sell => "loss",
            OrderSide::Buy => "entry",
        });
        body["stopPrice"] = serde_json::json!(format_decimal(stop));
        if self.is_futures() {
            body["stopPriceType"] = serde_json::json!("MP");
        }
        Ok(())
    }

    pub fn place_order(&self, request: &OrderRequest) -> Result<Order> {
        self.ensure_reduce_only_is_reducible(request)?;
        self.ensure_one_way()?;
        request.validate()?;
        let client_oid = request
            .client_order_id
            .clone()
            .unwrap_or_else(|| self.client_ids.next_id());
        let symbol_wire = if self.is_futures() {
            Self::futures_symbol(&request.symbol)
        } else {
            Self::wire_symbol(&request.symbol)
        };
        let mut body = serde_json::json!({
            "clientOid": client_oid,
            "side": side_str(request.side),
            "symbol": symbol_wire,
            "type": order_type_str(request.order_type),
            "size": format_decimal(request.quantity),
        });
        if let Some(price) = request.price {
            body["price"] = serde_json::json!(format_decimal(price));
        }
        if request.post_only {
            body["postOnly"] = serde_json::json!(true);
        }
        if let Some(tif) = tif_str(request.time_in_force) {
            body["timeInForce"] = serde_json::json!(tif);
        }
        if let Some(stp) = stp_flag(request.stp) {
            body["stp"] = serde_json::json!(stp);
        }
        if self.is_futures() {
            // Futures orders carry the per-order leverage; size is in contracts.
            body["leverage"] = serde_json::json!(self.leverage.to_string());
            if request.reduce_only {
                body["reduceOnly"] = serde_json::json!(true);
            }
        }
        // Spot serves trigger orders from their own path; futures takes them on
        // this one, with three extra fields. The body above is shared, so the
        // two cannot drift on everything that is not the trigger.
        let path = if request.order_type.is_trigger() {
            self.add_trigger(&mut body, request)?;
            if self.is_futures() {
                "/api/v1/orders"
            } else {
                "/api/v1/stop-order"
            }
        } else {
            "/api/v1/orders"
        };
        let data = self.signed_request(HttpMethod::Post, path, "", &body.to_string())?;
        let placed: PlaceResult = parse_json(data)?;
        Ok(Order {
            id: placed.order_id,
            client_order_id: Some(client_oid),
            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<()> {
        let path = format!("/api/v1/orders/{order_id}");
        self.signed_request(HttpMethod::Delete, &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> {
        let path = format!("/api/v1/orders/{order_id}");
        let data = self.signed_request(HttpMethod::Get, &path, "", "")?;
        let raw: RawOrder = parse_json(data)?;
        order_from_raw(symbol.clone(), &raw)
    }

    /// All 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>> {
        let mut query = "status=active".to_string();
        if let Some(s) = symbol {
            query.push_str("&symbol=");
            query.push_str(&Self::wire_symbol(s));
        }
        let data = self.signed_request(HttpMethod::Get, "/api/v1/orders", &query, "")?;
        let page: OrderPage = parse_json(data)?;
        page.items
            .iter()
            .map(|raw| {
                let sym = symbol.cloned().unwrap_or_else(|| {
                    raw.symbol
                        .parse()
                        .unwrap_or_else(|_| Symbol::new(&raw.symbol, ""))
                });
                order_from_raw(sym, raw)
            })
            .collect()
    }

    /// Trade-account balances.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn balances(&self) -> Result<Vec<Balance>> {
        let data = self.signed_request(HttpMethod::Get, "/api/v1/accounts", "type=trade", "")?;
        let list: Vec<RawAccount> = parse_json(data)?;
        Ok(list
            .iter()
            .map(|a| Balance {
                asset: a.currency.clone(),
                free: dec_or_zero(&a.available),
                locked: dec_or_zero(&a.holds),
            })
            .collect())
    }

    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_envelope(&response.body).map_err(|e| e.with_retry_after(response.retry_after()))
    }

    /// Sign with the `KC-API-*` headers (key version 2, HMAC-signed passphrase).
    fn signed_request(
        &self,
        method: HttpMethod,
        path: &str,
        query: &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 passphrase = creds
            .passphrase
            .as_deref()
            .ok_or(Error::InvalidCredentials("KuCoin requires a passphrase"))?;
        let timestamp = self.signed_now_ms().to_string();
        let endpoint = if query.is_empty() {
            path.to_string()
        } else {
            format!("{path}?{query}")
        };
        let prehash = format!("{timestamp}{}{endpoint}{body}", method.as_str());
        let signature = hmac_sha256_base64(creds.api_secret.as_bytes(), prehash.as_bytes());
        let signed_passphrase =
            hmac_sha256_base64(creds.api_secret.as_bytes(), passphrase.as_bytes());
        let url = format!("{}{endpoint}", self.rest_base);
        let mut request = HttpRequest::new(method, url)
            .with_header("KC-API-KEY", creds.api_key.clone())
            .with_header("KC-API-SIGN", signature)
            .with_header("KC-API-TIMESTAMP", timestamp)
            .with_header("KC-API-PASSPHRASE", signed_passphrase)
            .with_header("KC-API-KEY-VERSION", "2");
        if !body.is_empty() {
            request = request
                .with_header("Content-Type", "application/json")
                .with_body(body.to_string());
        }
        let response = self.http.execute(&request)?;
        unwrap_envelope(&response.body).map_err(|e| e.with_retry_after(response.retry_after()))
    }
}

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

fn unwrap_envelope(body: &str) -> Result<serde_json::Value> {
    let envelope: Envelope =
        serde_json::from_str(body).map_err(|e| Error::Deserialization(e.to_string()))?;
    if envelope.code != "200000" {
        return Err(map_error(&envelope.code, &envelope.msg));
    }
    Ok(envelope.data)
}

fn parse_json<T: for<'de> Deserialize<'de>>(value: serde_json::Value) -> Result<T> {
    serde_json::from_value(value).map_err(|e| Error::Deserialization(e.to_string()))
}

fn map_error(code: &str, msg: &str) -> Error {
    match code {
        "429000" => Error::RateLimited { retry_after: None },
        "400003" | "400004" | "400005" | "400006" | "400007" => Error::Auth(msg.to_string()),
        "200004" | "400100" => Error::InsufficientBalance,
        "400600" | "404000" => Error::NotFound(msg.to_string()),
        other => Error::Exchange {
            code: other.to_string(),
            message: msg.to_string(),
        },
    }
}

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

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

/// The KuCoin `stp` flag for a self-trade-prevention policy, or `None` to omit.
/// KuCoin cancels the oldest (`CO`), newest (`CN`) or both (`CB`) order.
/// The KuCoin `timeInForce` value. KuCoin carries it as its own field beside
/// `postOnly`, so the two never conflict; `Gtc` is KuCoin's default and is left
/// off the wire.
fn tif_str(tif: TimeInForce) -> Option<&'static str> {
    match tif {
        TimeInForce::Gtc => None,
        TimeInForce::Ioc => Some("IOC"),
        TimeInForce::Fok => Some("FOK"),
    }
}

fn stp_flag(stp: SelfTradePrevention) -> Option<&'static str> {
    match stp {
        SelfTradePrevention::None => None,
        SelfTradePrevention::ExpireMaker => Some("CO"),
        SelfTradePrevention::ExpireTaker => Some("CN"),
        SelfTradePrevention::ExpireBoth => Some("CB"),
    }
}

fn parse_side(raw: &str) -> Result<OrderSide> {
    match raw {
        "buy" => Ok(OrderSide::Buy),
        "sell" => Ok(OrderSide::Sell),
        other => Err(Error::Deserialization(format!("unknown side {other:?}"))),
    }
}

fn parse_order_type(raw: &str) -> Result<OrderType> {
    match raw {
        "market" => Ok(OrderType::Market),
        "limit" => Ok(OrderType::Limit),
        other => Err(Error::Deserialization(format!(
            "unknown order type {other:?}"
        ))),
    }
}

fn dec_or_zero(raw: &str) -> Decimal {
    crate::normalize::parse_opt_decimal(Some(raw))
        .ok()
        .flatten()
        .unwrap_or(Decimal::ZERO)
}

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

fn parse_levels(levels: &[[String; 2]]) -> Result<Vec<BookLevel>> {
    levels
        .iter()
        .map(|[price, qty]| {
            Ok(BookLevel {
                price: parse_decimal(price)?,
                quantity: parse_decimal(qty)?,
            })
        })
        .collect()
}

fn parse_kline_row(row: &[String]) -> Result<Candle> {
    // KuCoin candle: [time, open, close, high, low, volume, turnover].
    if row.len() < 6 {
        return Err(Error::Deserialization("kline row too short".to_string()));
    }
    let ts = row[0]
        .parse::<i64>()
        .map_err(|e| Error::Deserialization(format!("kline time not an integer: {e}")))?;
    let f = |i: usize| -> Result<f64> {
        row[i]
            .parse::<f64>()
            .map_err(|e| Error::Deserialization(format!("kline field not a number: {e}")))
    };
    // Note the KuCoin field order: open, close, high, low.
    Candle::new(f(1)?, f(3)?, f(4)?, f(2)?, f(5)?, ts)
        .map_err(|e| Error::Deserialization(e.to_string()))
}

fn order_from_raw(symbol: Symbol, raw: &RawOrder) -> Result<Order> {
    let status = if raw.is_active {
        if dec_or_zero(&raw.deal_size) > Decimal::ZERO {
            OrderStatus::PartiallyFilled
        } else {
            OrderStatus::New
        }
    } else if raw.cancel_exist {
        OrderStatus::Canceled
    } else {
        OrderStatus::Filled
    };
    Ok(Order {
        id: raw.id.clone(),
        client_order_id: (!raw.client_oid.is_empty()).then(|| raw.client_oid.clone()),
        symbol,
        side: parse_side(&raw.side)?,
        order_type: parse_order_type(&raw.order_type)?,
        status,
        quantity: parse_decimal(&raw.size)?,
        filled_quantity: dec_or_zero(&raw.deal_size),
        price: nonzero_decimal(&raw.price),
        average_price: None,
    })
}

fn field_str<'a>(value: &'a serde_json::Value, key: &str) -> Result<&'a str> {
    value
        .get(key)
        .and_then(serde_json::Value::as_str)
        .ok_or_else(|| Error::Deserialization(format!("missing string field {key:?}")))
}

fn opt_str<'a>(value: &'a serde_json::Value, key: &str) -> &'a str {
    value
        .get(key)
        .and_then(serde_json::Value::as_str)
        .unwrap_or("")
}

/// Read a `u64` that a venue may encode as either a JSON number or a string.
fn opt_u64(value: &serde_json::Value, key: &str) -> u64 {
    value
        .get(key)
        .and_then(|v| {
            v.as_u64()
                .or_else(|| v.as_str().and_then(|s| s.parse().ok()))
        })
        .unwrap_or(0)
}

fn parse_ws_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 = parse_decimal(
                pair.first()
                    .and_then(serde_json::Value::as_str)
                    .ok_or_else(|| Error::Deserialization("depth price missing".to_string()))?,
            )?;
            let quantity =
                parse_decimal(pair.get(1).and_then(serde_json::Value::as_str).ok_or_else(
                    || Error::Deserialization("depth quantity missing".to_string()),
                )?)?;
            Ok(BookLevel { price, quantity })
        })
        .collect()
}

/// An integer field, or zero. KuCoin's futures frames send sizes and stamps as
/// JSON numbers where its spot frames send strings.
fn opt_i64(value: &serde_json::Value, key: &str) -> i64 {
    value
        .get(key)
        .and_then(serde_json::Value::as_i64)
        .unwrap_or(0)
}

/// KuCoin Futures' WebSocket host.
const FUTURES_WS_URL: &str = "wss://ws-api-futures.kucoin.com/";

/// The futures topic that answers the same question as a spot one.
///
/// The two streams are not the same protocol with a prefix swapped -- the
/// payloads differ as well -- but the mapping of *what is asked for* is
/// one-to-one, so it lives in one place.
fn futures_topic(spot_topic: &str) -> &'static str {
    match spot_topic {
        "/market/match" => "/contractMarket/execution",
        "/market/level2" => "/contractMarket/level2",
        _ => "/contractMarket/ticker",
    }
}

/// KuCoin Futures stamps in nanoseconds where its spot stream uses
/// milliseconds. Reporting the raw figure would put every futures event about
/// 56,000 years into the future.
fn futures_millis(raw: i64) -> i64 {
    raw / 1_000_000
}

/// One level from `/contractMarket/level2`, which sends `"price,side,size"` as
/// a single string rather than a bids/asks object. A size of zero removes the
/// level, which is what the delta type already means.
fn futures_change(change: &str) -> Option<(BookLevel, OrderSide)> {
    let mut parts = change.split(',');
    let price = parse_decimal(parts.next()?.trim()).ok()?;
    let side = match parts.next()?.trim() {
        "buy" => OrderSide::Buy,
        "sell" => OrderSide::Sell,
        _ => return None,
    };
    let quantity = parse_decimal(parts.next()?.trim()).ok()?;
    Some((BookLevel { price, quantity }, side))
}

fn parse_ws_message(
    text: &str,
    resolve: &impl Fn(&str) -> Symbol,
    subscribed: &impl Fn(&str, DerivativesChannel) -> bool,
) -> Result<Option<Event>> {
    let value: serde_json::Value =
        serde_json::from_str(text).map_err(|e| Error::Deserialization(e.to_string()))?;
    if value.get("type").and_then(serde_json::Value::as_str) != Some("message") {
        return Ok(None); // ack/welcome/pong
    }
    let Some(topic) = value.get("topic").and_then(serde_json::Value::as_str) else {
        return Ok(None);
    };
    let wire = topic.rsplit(':').next().unwrap_or("");
    let symbol = resolve(wire);
    let null = serde_json::Value::Null;
    let data = value.get("data").unwrap_or(&null);

    // --- futures ------------------------------------------------------------
    if topic.starts_with("/contract/instrument:") {
        // One topic, two subjects. `funding.rate` carries no price, so a
        // funding print reports a mark of zero rather than one taken from the
        // other subject at a different moment.
        return Ok(
            match value.get("subject").and_then(serde_json::Value::as_str) {
                Some("funding.rate") if subscribed(wire, DerivativesChannel::Funding) => {
                    let rate = data
                        .get("fundingRate")
                        .and_then(serde_json::Value::as_f64)
                        .and_then(|r| parse_decimal(&r.to_string()).ok());
                    rate.map(|rate| {
                        Event::Derivatives(DerivativesFeed::Funding(FundingRate {
                            symbol,
                            rate,
                            mark_price: Decimal::ZERO,
                            timestamp: opt_i64(data, "timestamp"),
                        }))
                    })
                }
                Some("mark.index.price") if subscribed(wire, DerivativesChannel::MarkIndex) => {
                    let mark = data
                        .get("markPrice")
                        .and_then(serde_json::Value::as_f64)
                        .and_then(|p| parse_decimal(&p.to_string()).ok());
                    let index = data
                        .get("indexPrice")
                        .and_then(serde_json::Value::as_f64)
                        .and_then(|p| parse_decimal(&p.to_string()).ok());
                    match (mark, index) {
                        (Some(mark_price), Some(index_price)) => {
                            Some(Event::Derivatives(DerivativesFeed::MarkIndex(MarkIndex {
                                symbol,
                                mark_price,
                                index_price,
                                timestamp: opt_i64(data, "timestamp"),
                            })))
                        }
                        _ => None,
                    }
                }
                _ => None,
            },
        );
    } else if topic.starts_with("/contractMarket/execution:") {
        // `size` is an integer contract count and `ts` is in nanoseconds.
        return Ok(Some(Event::Trade(TradePrint {
            symbol,
            price: parse_decimal(field_str(data, "price")?)?,
            quantity: parse_decimal(&opt_i64(data, "size").to_string())?,
            aggressor: parse_side(field_str(data, "side")?)?,
            timestamp: futures_millis(opt_i64(data, "ts")),
        })));
    } else if topic.starts_with("/contractMarket/ticker:") {
        return Ok(Some(Event::Ticker(Ticker {
            symbol,
            last: parse_decimal(field_str(data, "price")?)?,
            bid: dec_or_zero(opt_str(data, "bestBidPrice")),
            ask: dec_or_zero(opt_str(data, "bestAskPrice")),
            // The futures ticker publishes no traded volume; inventing one from
            // the print size would be a different figure wearing the same name.
            volume: Decimal::ZERO,
            timestamp: futures_millis(opt_i64(data, "ts")),
        })));
    } else if topic.starts_with("/contractMarket/level2:") {
        // One level per frame, as `"price,side,size"`.
        let Some((level, side)) = futures_change(opt_str(data, "change")) else {
            return Ok(None);
        };
        let sequence = opt_i64(data, "sequence");
        let (bids, asks) = match side {
            OrderSide::Buy => (vec![level], Vec::new()),
            OrderSide::Sell => (Vec::new(), vec![level]),
        };
        return Ok(Some(Event::BookDelta(BookDelta {
            symbol,
            first_update_id: sequence.unsigned_abs(),
            final_update_id: sequence.unsigned_abs(),
            bids,
            asks,
            timestamp: opt_i64(data, "timestamp"),
        })));
    }

    if topic.starts_with("/market/match:") {
        Ok(Some(Event::Trade(TradePrint {
            symbol,
            price: parse_decimal(field_str(data, "price")?)?,
            quantity: parse_decimal(field_str(data, "size")?)?,
            aggressor: parse_side(field_str(data, "side")?)?,
            timestamp: opt_str(data, "time").parse().unwrap_or(0),
        })))
    } else if topic.starts_with("/market/ticker:") {
        Ok(Some(Event::Ticker(Ticker {
            symbol,
            last: parse_decimal(field_str(data, "price")?)?,
            bid: dec_or_zero(opt_str(data, "bestBid")),
            ask: dec_or_zero(opt_str(data, "bestAsk")),
            volume: dec_or_zero(opt_str(data, "size")),
            timestamp: data
                .get("time")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })))
    } else if topic.starts_with("/market/level2:") {
        let changes = data.get("changes");
        let bids = parse_ws_levels(changes.and_then(|c| c.get("bids")))?;
        let asks = parse_ws_levels(changes.and_then(|c| c.get("asks")))?;
        let update_id = opt_u64(data, "sequenceEnd");
        Ok(Some(Event::BookDelta(BookDelta {
            symbol,
            first_update_id: update_id,
            final_update_id: update_id,
            bids,
            asks,
            timestamp: data
                .get("time")
                .and_then(serde_json::Value::as_i64)
                .unwrap_or(0),
        })))
    } else if topic == "/spotMarket/tradeOrders" {
        Ok(Some(Event::OrderUpdate(ws_order_from_data(data)?)))
    } else if topic == "/contractMarket/tradeOrders" {
        Ok(Some(Event::OrderUpdate(futures_order_from_data(data)?)))
    } else if topic == "/contractAccount/wallet" {
        // One currency per frame, the way the spot balance channel sends it.
        // `holdBalance` is the margin already committed to open orders and
        // positions, which is what `locked` means everywhere else here.
        Ok(Some(Event::BalanceUpdate(vec![Balance {
            asset: field_str(data, "currency")?.to_string(),
            free: dec_or_zero(opt_str(data, "availableBalance")),
            locked: dec_or_zero(opt_str(data, "holdBalance")),
        }])))
    } else if topic == "/account/balance" {
        Ok(Some(Event::BalanceUpdate(vec![Balance {
            asset: field_str(data, "currency")?.to_string(),
            free: dec_or_zero(opt_str(data, "available")),
            locked: dec_or_zero(opt_str(data, "hold")),
        }])))
    } else {
        Ok(None)
    }
}

/// Turn a KuCoin **futures** contract symbol back into a [`Symbol`].
///
/// `XBTUSDTM` -> `BTC/USDT`: KuCoin spells Bitcoin `XBT` and marks the
/// perpetual with a trailing `M`. The private order stream carries this form
/// and is not subscribed per symbol, so there is no subscription to look the
/// symbol up in -- it has to be read back out of the wire name.
fn futures_symbol_from_wire(wire: &str) -> Symbol {
    let stem = wire.strip_suffix('M').unwrap_or(wire);
    // Longest first: `USDT` and `USDC` both end in a quote that is also a quote.
    let quote = ["USDT", "USDC", "USD"]
        .into_iter()
        .find(|q| stem.len() > q.len() && stem.ends_with(q));
    let Some(quote) = quote else {
        return Symbol::new(stem, "");
    };
    let base = &stem[..stem.len() - quote.len()];
    let base = if base == "XBT" { "BTC" } else { base };
    Symbol::new(base, quote)
}

/// Map a `/contractMarket/tradeOrders` status to an [`OrderStatus`].
///
/// The futures channel says less than the spot one: `done` covers both a fill
/// and a cancel, and only the event `type` separates them. Reading `status`
/// alone would report every cancelled order as filled.
fn futures_order_status(status: &str, event_type: &str) -> OrderStatus {
    match (status, event_type) {
        (_, "canceled") => OrderStatus::Canceled,
        ("match", _) => OrderStatus::PartiallyFilled,
        ("done", _) | (_, "filled") => OrderStatus::Filled,
        _ => OrderStatus::New,
    }
}

/// Build an [`Order`] from a `/contractMarket/tradeOrders` message payload.
fn futures_order_from_data(data: &serde_json::Value) -> Result<Order> {
    let client_oid = opt_str(data, "clientOid");
    Ok(Order {
        id: field_str(data, "orderId")?.to_string(),
        client_order_id: (!client_oid.is_empty()).then(|| client_oid.to_string()),
        symbol: futures_symbol_from_wire(field_str(data, "symbol")?),
        side: parse_side(field_str(data, "side")?)?,
        order_type: parse_order_type(field_str(data, "orderType")?)?,
        status: futures_order_status(opt_str(data, "status"), opt_str(data, "type")),
        // Futures sizes are contract counts and arrive as numbers, not strings.
        quantity: number_or_string(data, "size"),
        filled_quantity: number_or_string(data, "filledSize"),
        price: nonzero_decimal(opt_str(data, "price")),
        average_price: None,
    })
}

/// Read a field the futures stream may send as either a number or a string.
fn number_or_string(data: &serde_json::Value, key: &str) -> Decimal {
    data.get(key).map_or(Decimal::ZERO, |value| {
        value.as_str().map_or_else(
            || {
                value
                    .as_i64()
                    .and_then(|n| parse_decimal(&n.to_string()).ok())
                    .unwrap_or(Decimal::ZERO)
            },
            |text| parse_decimal(text).unwrap_or(Decimal::ZERO),
        )
    })
}

/// Map a `/spotMarket/tradeOrders` status (with the event `type` disambiguating a
/// finished order) to an [`OrderStatus`].
fn ws_order_status(status: &str, event_type: &str) -> OrderStatus {
    match status {
        "match" => OrderStatus::PartiallyFilled,
        "done" if event_type == "canceled" => OrderStatus::Canceled,
        "done" => OrderStatus::Filled,
        _ => OrderStatus::New,
    }
}

/// Build an [`Order`] from a `/spotMarket/tradeOrders` message payload.
fn ws_order_from_data(data: &serde_json::Value) -> Result<Order> {
    let wire = field_str(data, "symbol")?;
    let symbol = wire.parse().unwrap_or_else(|_| Symbol::new(wire, ""));
    let client_oid = opt_str(data, "clientOid");
    Ok(Order {
        id: field_str(data, "orderId")?.to_string(),
        client_order_id: (!client_oid.is_empty()).then(|| client_oid.to_string()),
        symbol,
        side: parse_side(field_str(data, "side")?)?,
        order_type: parse_order_type(field_str(data, "orderType")?)?,
        status: ws_order_status(opt_str(data, "status"), opt_str(data, "type")),
        quantity: parse_decimal(field_str(data, "size")?)?,
        filled_quantity: dec_or_zero(opt_str(data, "filledSize")),
        price: nonzero_decimal(opt_str(data, "price")),
        average_price: None,
    })
}

/// The bullet-private token negotiation response.
#[derive(Deserialize)]
struct BulletToken {
    token: String,
    #[serde(rename = "instanceServers", default)]
    instance_servers: Vec<BulletServer>,
}

#[derive(Deserialize)]
struct BulletServer {
    endpoint: String,
}

#[derive(Deserialize)]
struct Envelope {
    code: String,
    #[serde(default)]
    msg: String,
    #[serde(default)]
    data: serde_json::Value,
}

#[derive(Deserialize)]
struct RawStats {
    /// Verified against the live endpoint: KuCoin stamps this payload.
    #[serde(default)]
    time: i64,
    last: String,
    buy: String,
    sell: String,
    vol: String,
}

#[derive(Deserialize)]
struct RawDepth {
    /// Verified against the live endpoint: KuCoin stamps this payload.
    #[serde(default)]
    time: i64,
    #[serde(default)]
    sequence: String,
    bids: Vec<[String; 2]>,
    asks: Vec<[String; 2]>,
}

#[derive(Deserialize)]
struct PlaceResult {
    #[serde(rename = "orderId", default)]
    order_id: String,
}

#[derive(Deserialize)]
struct OrderPage {
    items: Vec<RawOrder>,
}

#[derive(Deserialize)]
struct RawOrder {
    id: String,
    #[serde(rename = "clientOid", default)]
    client_oid: String,
    #[serde(default)]
    symbol: String,
    side: String,
    #[serde(rename = "type")]
    order_type: String,
    size: String,
    #[serde(rename = "dealSize", default)]
    deal_size: String,
    #[serde(default)]
    price: String,
    #[serde(rename = "isActive", default)]
    is_active: bool,
    #[serde(rename = "cancelExist", default)]
    cancel_exist: bool,
}

#[derive(Deserialize)]
struct RawAccount {
    currency: String,
    #[serde(default)]
    available: String,
    #[serde(default)]
    holds: String,
}

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

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

impl KuCoin {
    /// Open futures positions (`/api/v1/positions` on the futures host); flat
    /// positions are omitted. Quantities are in **contracts**, KuCoin's native
    /// futures unit.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn positions(&self, symbol: Option<&Symbol>) -> Result<Vec<Position>> {
        let data = self.signed_request(HttpMethod::Get, "/api/v1/positions", "", "")?;
        let list: Vec<RawKuPosition> = parse_json(data)?;
        list.iter()
            .filter(|p| symbol.is_none_or(|s| p.symbol == Self::futures_symbol(s)))
            .filter_map(parse_ku_position)
            .collect()
    }

    /// Record the leverage applied to subsequent futures orders. KuCoin has no
    /// standalone set-leverage endpoint — leverage is a per-order field — so this
    /// stores it locally rather than issuing a request.
    ///
    /// # Errors
    /// Never errors; the signature matches the [`Derivatives`] trait.
    pub fn set_leverage(&mut self, _symbol: &Symbol, leverage: u32) -> Result<()> {
        self.leverage = leverage.max(1);
        Ok(())
    }

    /// Set the margin mode for `symbol` (`/api/v1/position/changeMarginMode`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the change is rejected or the request fails.
    pub fn set_margin_mode(&self, symbol: &Symbol, mode: MarginMode) -> Result<()> {
        let margin = match mode {
            MarginMode::Cross => "CROSS",
            MarginMode::Isolated => "ISOLATED",
        };
        let body = serde_json::json!({
            "symbol": Self::futures_symbol(symbol),
            "marginMode": margin,
        });
        self.signed_request(
            HttpMethod::Post,
            "/api/v1/position/changeMarginMode",
            "",
            &body.to_string(),
        )?;
        Ok(())
    }

    /// Flatten the open position in `symbol` with a reduce-only market order
    /// (size in contracts).
    ///
    /// # 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> {
        self.ensure_one_way()?;
        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)
    }

    /// Place several spot orders on one symbol in one request
    /// (`/api/v1/orders/multi`). KuCoin returns the results in request order, so
    /// each element maps to its request's own [`Result`].
    ///
    /// # Errors
    /// Returns an [`Error`] if the batch request itself fails, or if called on a
    /// futures client (multi is a spot 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("KuCoin"));
        }
        for request in requests {
            self.ensure_reduce_only_is_reducible(request)?;
        }
        self.ensure_one_way()?;
        if requests.is_empty() {
            return Ok(Vec::new());
        }
        if self.is_futures() {
            return Err(Error::Exchange {
                code: "unsupported".to_string(),
                message: "KuCoin multi-order is a spot endpoint".to_string(),
            });
        }
        let wire = Self::wire_symbol(&requests[0].symbol);
        let order_list: Vec<serde_json::Value> = requests
            .iter()
            .map(|r| {
                let coid = r
                    .client_order_id
                    .clone()
                    .unwrap_or_else(|| self.client_ids.next_id());
                let mut o = serde_json::json!({
                    "clientOid": coid,
                    "side": side_str(r.side),
                    "type": order_type_str(r.order_type),
                    "size": format_decimal(r.quantity),
                });
                if let Some(price) = r.price {
                    o["price"] = serde_json::json!(format_decimal(price));
                }
                if r.post_only {
                    o["postOnly"] = serde_json::json!(true);
                }
                if let Some(tif) = tif_str(r.time_in_force) {
                    o["timeInForce"] = serde_json::json!(tif);
                }
                if let Some(stp) = stp_flag(r.stp) {
                    o["stp"] = serde_json::json!(stp);
                }
                o
            })
            .collect();
        let body = serde_json::json!({ "symbol": wire, "orderList": order_list });
        let data = self.signed_request(
            HttpMethod::Post,
            "/api/v1/orders/multi",
            "",
            &body.to_string(),
        )?;
        let batch: MultiBatch = parse_json(data)?;
        Ok(requests
            .iter()
            .zip(batch.data)
            .map(|(req, res)| {
                if res.order_id.is_empty() || res.status == "fail" {
                    return Err(Error::OrderRejected {
                        code: "batch".to_string(),
                        message: if res.fail_msg.is_empty() {
                            "order rejected in batch".to_string()
                        } else {
                            res.fail_msg
                        },
                    });
                }
                Ok(Order {
                    id: res.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. KuCoin has no batch-cancel-by-id endpoint, so
    /// the ids are cancelled sequentially.
    ///
    /// # Errors
    /// Returns an [`Error`] if any cancel fails.
    pub fn cancel_batch(&self, symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        for id in order_ids {
            self.cancel_order(symbol, id)?;
        }
        Ok(())
    }

    /// Place a one-cancels-other bracket (`/api/v3/oco/order`). KuCoin models an
    /// OCO as one order-list, so the returned vector holds one order carrying the
    /// list id.
    ///
    /// # Errors
    /// Returns an [`Error`] if the OCO is invalid or rejected.
    pub fn place_oco(&self, request: &OcoRequest) -> Result<Vec<Order>> {
        self.ensure_one_way()?;
        request.validate()?;
        let client_oid = request
            .client_order_id
            .clone()
            .unwrap_or_else(|| self.client_ids.next_id());
        let stop_limit = request.stop_limit_price.unwrap_or(request.stop_price);
        let body = serde_json::json!({
            "symbol": Self::wire_symbol(&request.symbol),
            "side": side_str(request.side),
            "size": format_decimal(request.quantity),
            "price": format_decimal(request.price),
            "stopPrice": format_decimal(request.stop_price),
            "limitPrice": format_decimal(stop_limit),
            "clientOid": client_oid,
        });
        let data =
            self.signed_request(HttpMethod::Post, "/api/v3/oco/order", "", &body.to_string())?;
        let placed: PlaceResult = parse_json(data)?;
        Ok(vec![Order {
            id: placed.order_id,
            client_order_id: Some(client_oid),
            symbol: request.symbol.clone(),
            side: request.side,
            order_type: OrderType::StopLimit,
            status: OrderStatus::New,
            quantity: request.quantity,
            filled_quantity: Decimal::ZERO,
            price: Some(request.price),
            average_price: None,
        }])
    }
}

impl AdvancedOrders for KuCoin {
    /// KuCoin 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: "KuCoin has no in-place amend; cancel and re-place the order".to_string(),
        })
    }
    fn place_batch(&mut self, requests: &[OrderRequest]) -> Result<Vec<Result<Order>>> {
        KuCoin::place_batch(self, requests)
    }
    fn cancel_batch(&mut self, symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        KuCoin::cancel_batch(self, symbol, order_ids)
    }
    fn place_oco(&mut self, request: &OcoRequest) -> Result<Vec<Order>> {
        KuCoin::place_oco(self, request)
    }
}

#[derive(Deserialize)]
struct MultiBatch {
    #[serde(default)]
    data: Vec<MultiResult>,
}

#[derive(Deserialize)]
struct MultiResult {
    #[serde(rename = "orderId", default)]
    order_id: String,
    #[serde(default)]
    status: String,
    #[serde(rename = "failMsg", default)]
    fail_msg: String,
}

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

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

impl WsExecution for KuCoin {
    /// KuCoin exposes no WebSocket order-entry API — its WebSocket surface is
    /// subscription-only (market data plus the private `tradeOrders`/`balance`
    /// 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: "KuCoin has no WebSocket order-entry API; place orders over \
                      REST (POST /api/v1/orders)"
                .to_string(),
        })
    }

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

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

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

/// Reconstruct a canonical [`Symbol`] from a KuCoin futures contract symbol
/// (`XBTUSDTM` -> `BTC/USDT`): drop the trailing `M`, split off a known quote
/// suffix, and map `XBT` back to `BTC`.
fn symbol_from_futures(contract: &str) -> Symbol {
    const QUOTES: &[&str] = &["USDT", "USDC", "USD"];
    let stripped = contract.strip_suffix('M').unwrap_or(contract);
    for quote in QUOTES {
        if let Some(base) = stripped.strip_suffix(quote) {
            if !base.is_empty() {
                let base = if base == "XBT" { "BTC" } else { base };
                return Symbol::new(base, *quote);
            }
        }
    }
    Symbol::new(stripped, "")
}

#[derive(Deserialize)]
struct RawKuPosition {
    symbol: String,
    #[serde(rename = "currentQty")]
    current_qty: f64,
    #[serde(rename = "avgEntryPrice", default)]
    avg_entry_price: f64,
    #[serde(rename = "markPrice", default)]
    mark_price: f64,
    #[serde(rename = "realLeverage", default)]
    real_leverage: f64,
    #[serde(rename = "unrealisedPnl", default)]
    unrealised_pnl: f64,
    #[serde(rename = "crossMode", default)]
    cross_mode: bool,
}

fn parse_ku_position(raw: &RawKuPosition) -> Option<Result<Position>> {
    if raw.current_qty == 0.0 {
        return None; // flat position
    }
    let side = if raw.current_qty < 0.0 {
        PositionSide::Short
    } else {
        PositionSide::Long
    };
    let dec = |value: f64| Decimal::from_f64(value).unwrap_or_default();
    Some(Ok(Position {
        symbol: symbol_from_futures(&raw.symbol),
        side,
        quantity: dec(raw.current_qty.abs()),
        entry_price: dec(raw.avg_entry_price),
        mark_price: dec(raw.mark_price),
        leverage: dec(raw.real_leverage),
        unrealized_pnl: dec(raw.unrealised_pnl),
        margin_mode: if raw.cross_mode {
            MarginMode::Cross
        } else {
            MarginMode::Isolated
        },
    }))
}

#[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")
    }

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

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

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

        // A contract with no multiplier: the base-currency figure cannot be
        // derived, and guessing one would be a fabricated size.
        http.push_json(
            200,
            r#"{"code":"200000","data":{"symbol":"XBTUSDTM","openInterest":"100"}}"#,
        );
        assert!(matches!(
            KuCoin::open_interest(&kucoin, &symbol()),
            Err(Error::Deserialization(_))
        ));
    }

    /// Frames the futures parser cannot use are dropped.
    #[test]
    fn kucoin_drops_futures_frames_it_cannot_read() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![
            // A book change that is not `price,side,size`.
            Ok(Some(
                r#"{"topic":"/contractMarket/level2:XBTUSDTM","type":"message",
                "subject":"level2","data":{"sequence":1,"change":"nonsense"}}"#
                    .to_string(),
            )),
            // A change naming a side the venue does not use.
            Ok(Some(
                r#"{"topic":"/contractMarket/level2:XBTUSDTM","type":"message",
                "subject":"level2","data":{"sequence":1,"change":"1.0,sideways,2"}}"#
                    .to_string(),
            )),
            // An instrument frame with an unknown subject.
            Ok(Some(
                r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
                "subject":"something.else","data":{}}"#
                    .to_string(),
            )),
        ]);
        KuCoin::subscribe_book(&mut kucoin, &symbol()).unwrap();
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::MarkIndex)
            .unwrap();

        assert_eq!(KuCoin::poll_events(&mut kucoin), Vec::new());
    }

    /// An instrument frame missing its prices is dropped rather than half-read.
    #[test]
    fn kucoin_drops_an_instrument_frame_without_prices() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![
            Ok(Some(
                r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
                "subject":"mark.index.price","data":{"markPrice":1.0}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
                "subject":"funding.rate","data":{"timestamp":1}}"#
                    .to_string(),
            )),
        ]);
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::MarkIndex)
            .unwrap();
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::Funding).unwrap();

        assert_eq!(KuCoin::poll_events(&mut kucoin), Vec::new());
    }

    /// Both channels arrive on one topic, under two subjects.
    #[test]
    fn the_instrument_topic_answers_funding_and_mark_index() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![
            Ok(Some(
                r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
                "subject":"mark.index.price","data":{"markPrice":81316.40,
                "indexPrice":81334.48,"granularity":1000,"timestamp":1788462351000}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
                "subject":"funding.rate","data":{"period":1,"granularity":60000,
                "fundingRate":0.000021,"timestamp":1788464640000}}"#
                    .to_string(),
            )),
        ]);
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::MarkIndex)
            .unwrap();
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::Funding).unwrap();
        // One subscription frame: the second channel is on the same topic.
        assert!(ws.sent()[0].contains("/contract/instrument:XBTUSDTM"));

        let events = KuCoin::poll_events(&mut kucoin);
        let Event::Derivatives(DerivativesFeed::MarkIndex(mark)) = &events[0] else {
            panic!("expected mark/index, got {:?}", events[0]);
        };
        assert_eq!(mark.mark_price, dec!(81316.40));
        assert_eq!(mark.index_price, dec!(81334.48));
        assert_eq!(mark.timestamp, 1_788_462_351_000);

        let Event::Derivatives(DerivativesFeed::Funding(funding)) = &events[1] else {
            panic!("expected funding, got {:?}", events[1]);
        };
        assert_eq!(funding.rate, dec!(0.000021));
        // The funding subject carries no price, so the mark is zero rather than
        // one borrowed from the other subject at another moment.
        assert_eq!(funding.mark_price, Decimal::ZERO);
    }

    /// A frame for a channel that was not subscribed to is not emitted.
    #[test]
    fn an_unsubscribed_instrument_subject_is_dropped() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"topic":"/contract/instrument:XBTUSDTM","type":"message",
            "subject":"funding.rate","data":{"fundingRate":0.000021,
            "timestamp":1788464640000}}"#
                .to_string(),
        ))]);
        KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::MarkIndex)
            .unwrap();

        assert_eq!(KuCoin::poll_events(&mut kucoin), Vec::new());
    }

    /// Open interest is in contracts; the multiplier arrives in the same reply,
    /// so the base-currency figure is arithmetic on one reading.
    #[test]
    fn open_interest_multiplies_contracts_by_the_contract_size() {
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let kucoin = KuCoin::with_http(Box::new(ArcTransport(Arc::clone(&http))), &opts);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"symbol":"XBTUSDTM","baseCurrency":"XBT",
            "multiplier":0.001,"openInterest":"16731943","lotSize":1}}"#,
        );
        let open_interest = KuCoin::open_interest(&kucoin, &symbol()).unwrap();
        assert_eq!(open_interest.open_interest, dec!(16731.943));
        assert!(http.recorded_requests()[0]
            .url
            .contains("/contracts/XBTUSDTM"));
    }

    /// KuCoin publishes neither forced orders nor a positioning figure.
    #[test]
    fn kucoin_refuses_what_it_does_not_publish() {
        let (mut kucoin, ws) = futures_ws_client();
        let err =
            KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), DerivativesChannel::Liquidations)
                .expect_err("KuCoin has no public liquidation stream");
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert_eq!(ws.sent(), Vec::<String>::new());

        let (kucoin, _) = futures_ws_client();
        assert!(KuCoin::long_short_ratio(&kucoin, &symbol()).is_err());
    }

    #[test]
    fn a_spot_client_refuses_the_kucoin_derivatives_surface() {
        let ws = Arc::new(MockWsTransport::new());
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut kucoin = KuCoin::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!(KuCoin::subscribe_derivatives(&mut kucoin, &symbol(), channel).is_err());
        }
        assert!(KuCoin::open_interest(&kucoin, &symbol()).is_err());
    }

    /// Host, topic and symbol all follow the market.
    #[test]
    fn a_futures_client_subscribes_to_contract_topics() {
        let (mut kucoin, ws) = futures_ws_client();
        KuCoin::subscribe_trades(&mut kucoin, &symbol()).unwrap();
        KuCoin::subscribe_book(&mut kucoin, &symbol()).unwrap();
        KuCoin::subscribe_ticker(&mut kucoin, &symbol()).unwrap();

        assert_eq!(ws.connected_urls()[0], "wss://ws-api-futures.kucoin.com/");
        assert!(ws.sent()[0].contains("/contractMarket/execution:XBTUSDTM"));
        assert!(ws.sent()[1].contains("/contractMarket/level2:XBTUSDTM"));
        assert!(ws.sent()[2].contains("/contractMarket/ticker:XBTUSDTM"));
        assert!(ws.sent().iter().all(|f| !f.contains("/market/")));
    }

    /// KuCoin Futures stamps in **nanoseconds**.
    ///
    /// Carrying the figure through unchanged would put every futures event
    /// roughly 56,000 years into the future, and every staleness check with it.
    #[test]
    fn a_futures_trade_frame_converts_its_nanosecond_stamp() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"topic":"/contractMarket/execution:XBTUSDTM","type":"message",
            "subject":"match","sn":1942782214230,"data":{"symbol":"XBTUSDTM",
            "sequence":1942782214230,"side":"buy","size":104,"price":"81303.2",
            "takerOrderId":"485067687312056321","tradeId":"1942782214230",
            "ts":1788462350779000000}}"#
                .to_string(),
        ))]);
        KuCoin::subscribe_trades(&mut kucoin, &symbol()).unwrap();

        let events = KuCoin::poll_events(&mut kucoin);
        let Event::Trade(trade) = &events[0] else {
            panic!("expected a trade, got {:?}", events[0]);
        };
        assert_eq!(trade.price, dec!(81303.2));
        // Contracts, which is the unit the futures order path takes.
        assert_eq!(trade.quantity, dec!(104));
        assert_eq!(trade.aggressor, OrderSide::Buy);
        assert_eq!(trade.timestamp, 1_788_462_350_779);
    }

    /// The futures book sends one level as `"price,side,size"`, not an object.
    #[test]
    fn a_futures_book_change_is_one_csv_level() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![
            Ok(Some(
                r#"{"topic":"/contractMarket/level2:XBTUSDTM","type":"message",
                "subject":"level2","sn":1747328682771,"data":{"sequence":1747328682771,
                "change":"81385.8,sell,0","timestamp":1788462349758}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"topic":"/contractMarket/level2:XBTUSDTM","type":"message",
                "subject":"level2","sn":1747328682772,"data":{"sequence":1747328682772,
                "change":"81300.1,buy,42","timestamp":1788462349759}}"#
                    .to_string(),
            )),
        ]);
        KuCoin::subscribe_book(&mut kucoin, &symbol()).unwrap();

        let events = KuCoin::poll_events(&mut kucoin);
        let Event::BookDelta(removed) = &events[0] else {
            panic!("expected a delta, got {:?}", events[0]);
        };
        assert_eq!(removed.bids, Vec::new());
        assert_eq!(removed.asks[0].price, dec!(81385.8));
        // A size of zero removes the level, which the delta type already means.
        assert_eq!(removed.asks[0].quantity, Decimal::ZERO);
        assert_eq!(removed.timestamp, 1_788_462_349_758);

        let Event::BookDelta(added) = &events[1] else {
            panic!("expected a delta, got {:?}", events[1]);
        };
        assert_eq!(added.asks, Vec::new());
        assert_eq!(added.bids[0].quantity, dec!(42));
    }

    /// The futures ticker carries a last price and a top of book, and no volume
    /// — so volume stays zero rather than borrowing the print size.
    #[test]
    fn a_futures_ticker_parses() {
        let (mut kucoin, ws) = futures_ws_client();
        ws.push_connection(vec![Ok(Some(
            r#"{"topic":"/contractMarket/ticker:XBTUSDTM","type":"message","subject":"ticker",
            "sn":1942783616429,"data":{"symbol":"XBTUSDTM","sequence":1942783616429,
            "side":"buy","size":1,"price":"81484.4","bestBidSize":4775,
            "bestBidPrice":"81484.3","bestAskPrice":"81487.7","bestAskSize":68,
            "ts":1788463466281000000}}"#
                .to_string(),
        ))]);
        KuCoin::subscribe_ticker(&mut kucoin, &symbol()).unwrap();

        let events = KuCoin::poll_events(&mut kucoin);
        let Event::Ticker(ticker) = &events[0] else {
            panic!("expected a ticker, got {:?}", events[0]);
        };
        assert_eq!(ticker.last, dec!(81484.4));
        assert_eq!(ticker.bid, dec!(81484.3));
        assert_eq!(ticker.ask, dec!(81487.7));
        assert_eq!(ticker.volume, Decimal::ZERO);
        assert_eq!(ticker.timestamp, 1_788_463_466_281);
    }

    /// The spot client keeps its own host and topics.
    #[test]
    fn a_spot_client_still_subscribes_to_market_topics() {
        let ws = Arc::new(MockWsTransport::new());
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut kucoin = KuCoin::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        KuCoin::subscribe_trades(&mut kucoin, &symbol()).unwrap();

        assert_eq!(ws.connected_urls()[0], "wss://ws-api-spot.kucoin.com/");
        assert!(ws.sent()[0].contains("/market/match:BTC-USDT"));
    }

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

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

    /// The same client on the futures host, which is a separate account with
    /// separate topics.
    fn signed_futures_ws_client(
        now_ms: i64,
    ) -> (KuCoin, Arc<MockHttpTransport>, Arc<MockWsTransport>) {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let kucoin = KuCoin::with_credentials(
            Box::new(ArcTransport(Arc::clone(&http))),
            &opts,
            Credentials::new("APIKEY", "SECRET").with_passphrase("PASS"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))))
        .with_clock(Box::new(move || now_ms));
        (kucoin, http, ws)
    }
    /// A futures client watches the futures account, not the spot one.
    ///
    /// Its own orders never appear in the spot account, so subscribing there is
    /// waiting for a fill that cannot arrive. The negotiation is the same call
    /// on the futures host, which was verified live: `bullet-public` on
    /// `api-futures.kucoin.com` answers with an instance server.
    #[test]
    fn a_futures_client_subscribes_to_the_contract_account_not_the_spot_one() {
        let (mut kucoin, http, ws) = signed_futures_ws_client(1_700_000_000_000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"t","instanceServers":[{"endpoint":"wss://ws-api-futures.kucoin.com/"}]}}"#,
        );
        kucoin.subscribe_user_data().unwrap();

        let negotiated = &http.recorded_requests()[0];
        assert!(
            negotiated.url.starts_with("https://api-futures.kucoin.com"),
            "{}",
            negotiated.url
        );
        assert!(negotiated.url.contains("/api/v1/bullet-private"));
        let sent = ws.sent();
        assert!(
            sent[0].contains("/contractMarket/tradeOrders"),
            "{}",
            sent[0]
        );
        assert!(sent[1].contains("/contractAccount/wallet"), "{}", sent[1]);
        assert!(
            !sent.iter().any(|frame| frame.contains("/spotMarket/")),
            "a futures client subscribed to a spot topic: {sent:?}"
        );
    }

    /// The futures order frame names the contract, not the pair, and says less
    /// than the spot one: `done` covers a fill and a cancel alike.
    #[test]
    fn a_contract_order_frame_becomes_an_order_update() {
        let (mut kucoin, http, ws) = signed_futures_ws_client(1_700_000_000_000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"t","instanceServers":[{"endpoint":"wss://ws-api-futures.kucoin.com/"}]}}"#,
        );
        ws.push_connection(vec![Ok(Some(
            r#"{"type":"message","topic":"/contractMarket/tradeOrders","subject":"symbolOrderChange",
                "data":{"orderId":"f-1","symbol":"XBTUSDTM","type":"canceled","status":"done",
                        "side":"sell","orderType":"limit","size":3,"filledSize":1,
                        "price":"19000","clientOid":"mine"}}"#
                .to_string(),
        ))]);
        kucoin.subscribe_user_data().unwrap();

        let events = kucoin.poll_events();
        let Some(Event::OrderUpdate(order)) = events.first() else {
            panic!("expected an order update, got {events:?}");
        };
        assert_eq!(order.id, "f-1");
        // `XBTUSDTM` is BTC/USDT: KuCoin spells Bitcoin XBT and marks the
        // perpetual with a trailing M.
        assert_eq!(order.symbol, Symbol::new("BTC", "USDT"));
        // `done` with a `canceled` type is a cancel; reading the status alone
        // would report it as filled.
        assert_eq!(order.status, OrderStatus::Canceled);
        assert_eq!(order.quantity, dec!(3));
        assert_eq!(order.filled_quantity, dec!(1));
        assert_eq!(order.client_order_id.as_deref(), Some("mine"));
    }

    #[test]
    fn a_contract_wallet_frame_becomes_a_balance_update() {
        let (mut kucoin, http, ws) = signed_futures_ws_client(1_700_000_000_000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"t","instanceServers":[{"endpoint":"wss://ws-api-futures.kucoin.com/"}]}}"#,
        );
        ws.push_connection(vec![Ok(Some(
            r#"{"type":"message","topic":"/contractAccount/wallet","subject":"availableBalance.change",
                "data":{"currency":"USDT","availableBalance":"120.5","holdBalance":"9.5"}}"#
                .to_string(),
        ))]);
        kucoin.subscribe_user_data().unwrap();

        let events = kucoin.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!(120.5));
        // Margin committed to open orders and positions is locked, not free.
        assert_eq!(balances[0].locked, dec!(9.5));
    }

    /// A futures contract name maps back to the pair it was built from, and an
    /// unknown one is not silently turned into a pair that does not exist.
    #[test]
    fn a_contract_name_maps_back_to_its_pair() {
        assert_eq!(
            futures_symbol_from_wire("XBTUSDTM"),
            Symbol::new("BTC", "USDT")
        );
        assert_eq!(
            futures_symbol_from_wire("ETHUSDCM"),
            Symbol::new("ETH", "USDC")
        );
        assert_eq!(
            futures_symbol_from_wire("XBTUSDM"),
            Symbol::new("BTC", "USD")
        );
        // Round-trips with the name the client sends.
        assert_eq!(
            futures_symbol_from_wire(&KuCoin::futures_symbol(&Symbol::new("SOL", "USDT"))),
            Symbol::new("SOL", "USDT")
        );
        // Nothing recognisable: kept whole rather than split at a guess.
        assert_eq!(futures_symbol_from_wire("WEIRD"), Symbol::new("WEIRD", ""));
    }

    #[test]
    fn subscribe_user_data_negotiates_token_and_streams_orders_and_balance() {
        let (mut kucoin, http, ws) = signed_ws_client(1000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"tok","instanceServers":[
            {"endpoint":"wss://push.kucoin.com/endpoint"}]}}"#,
        );
        ws.push_connection(vec![
            Ok(Some(r#"{"type":"ack","id":"1000"}"#.to_string())),
            Ok(Some(
                r#"{"type":"message","topic":"/spotMarket/tradeOrders","subject":"orderChange",
                "data":{"symbol":"BTC-USDT","orderId":"55","clientOid":"my","side":"buy",
                "orderType":"limit","type":"filled","status":"done","size":"1","filledSize":"1",
                "price":"100"}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"type":"message","topic":"/account/balance","subject":"account.balance",
                "data":{"currency":"USDT","available":"900","hold":"50","total":"950"}}"#
                    .to_string(),
            )),
        ]);
        kucoin.subscribe_user_data().unwrap();

        let reqs = http.recorded_requests();
        assert!(reqs[0].url.contains("/api/v1/bullet-private"));
        assert_eq!(reqs[0].method, HttpMethod::Post);
        assert_eq!(
            ws.connected_urls()[0],
            "wss://push.kucoin.com/endpoint?token=tok&connectId=1000"
        );
        assert!(ws.sent()[0].contains("/spotMarket/tradeOrders"));
        assert!(ws.sent()[1].contains("/account/balance"));

        let events = kucoin.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));
        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_a_passphrase() {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut kucoin = KuCoin::with_credentials(
            Box::new(ArcTransport(http)),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        assert!(matches!(
            kucoin.subscribe_user_data().unwrap_err(),
            Error::InvalidCredentials(_)
        ));
    }

    #[test]
    fn keepalive_user_data_pings_the_private_stream() {
        let (mut kucoin, http, ws) = signed_ws_client(1000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"tok","instanceServers":[
            {"endpoint":"wss://push.kucoin.com/endpoint"}]}}"#,
        );
        ws.push_connection(vec![]);
        kucoin.subscribe_user_data().unwrap();
        kucoin.keepalive_user_data().unwrap();
        assert!(ws.sent().iter().any(|f| f.contains(r#""type":"ping""#)));
    }

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

    #[test]
    fn dropped_user_data_stream_reconnects_with_a_fresh_bullet_token() {
        let (mut kucoin, http, ws) = signed_ws_client(1000);
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"tok-1","instanceServers":[
            {"endpoint":"wss://push.kucoin.com/endpoint"}]}}"#,
        );
        // 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![]);
        kucoin.subscribe_user_data().unwrap();
        // The reconnect re-negotiates a fresh bullet token over REST.
        http.push_json(
            200,
            r#"{"code":"200000","data":{"token":"tok-2","instanceServers":[
            {"endpoint":"wss://push.kucoin.com/endpoint"}]}}"#,
        );

        let events = kucoin.poll_events();
        assert!(events.contains(&Event::Disconnected));
        assert!(events.contains(&Event::Reconnected));
        // Two bullet-private POSTs (initial + reconnect) and a second WS connection.
        let bullets = http
            .recorded_requests()
            .into_iter()
            .filter(|r| r.url.contains("/api/v1/bullet-private"))
            .count();
        assert_eq!(bullets, 2);
        assert_eq!(ws.connected_urls().len(), 2);
        assert!(ws.connected_urls()[1].contains("token=tok-2"));
    }

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

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

    /// Spot triggers go to their own path; `stop` says which way the market has
    /// to cross the trigger and KuCoin will not infer it.
    ///
    /// `loss` fires when the market moves against the side -- down for a sell --
    /// which is what a stop-loss means. `entry` fires the other way, which is a
    /// breakout entry: a different order that happens to use the same price.
    #[test]
    fn a_spot_trigger_goes_to_the_stop_order_path_with_its_direction() {
        for (side, expected) in [(OrderSide::Sell, "loss"), (OrderSide::Buy, "entry")] {
            let (kucoin, mock) = signed_client(1_700_000_000_000);
            mock.push_json(200, r#"{"code":"200000","data":{"orderId":"s-1"}}"#);
            let request = OrderRequest {
                order_type: OrderType::StopMarket,
                stop_price: Some(dec!(19000)),
                side,
                ..OrderRequest::market_sell(symbol(), dec!(1))
            };
            KuCoin::place_order(&kucoin, &request).unwrap();

            let sent = &mock.recorded_requests()[0];
            assert!(sent.url.contains("/api/v1/stop-order"), "{}", sent.url);
            let body = sent.body.clone().unwrap();
            assert!(body.contains(r#""stopPrice":"19000""#), "{body}");
            assert!(
                body.contains(&format!(r#""stop":"{expected}""#)),
                "a {side:?} stop fires the other way: {body}"
            );
            // Spot has no mark price, so it takes no `stopPriceType`.
            assert!(!body.contains("stopPriceType"), "{body}");
        }
    }

    /// Futures takes the trigger on the ordinary order path, and names the mark
    /// as the price that arms it -- the price a position is liquidated against.
    #[test]
    fn a_futures_trigger_stays_on_the_order_path_and_watches_the_mark() {
        let (kucoin, mock) = signed_futures_client(1_700_000_000_000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"f-1"}}"#);
        let request = OrderRequest {
            order_type: OrderType::StopLimit,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::limit_sell(symbol(), dec!(1), dec!(18900))
        };
        KuCoin::place_order(&kucoin, &request).unwrap();

        let sent = &mock.recorded_requests()[0];
        assert!(sent.url.contains("/api/v1/orders"), "{}", sent.url);
        assert!(!sent.url.contains("stop-order"), "{}", sent.url);
        let body = sent.body.clone().unwrap();
        assert!(body.contains(r#""stopPriceType":"MP""#), "{body}");
        assert!(body.contains(r#""price":"18900""#), "{body}");
    }

    #[test]
    fn a_kucoin_trigger_without_a_stop_price_is_refused() {
        let (kucoin, mock) = signed_client(1_700_000_000_000);
        let request = OrderRequest {
            order_type: OrderType::StopMarket,
            stop_price: None,
            ..OrderRequest::market_sell(symbol(), dec!(1))
        };
        assert!(KuCoin::place_order(&kucoin, &request).is_err());
        assert_eq!(mock.recorded_requests(), Vec::new());
    }
    fn hedged_futures_client(now_ms: i64) -> (KuCoin, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let mut opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        opts.position_mode = PositionMode::Hedge;
        let kucoin = KuCoin::with_credentials(
            Box::new(ArcTransport(Arc::clone(&mock))),
            &opts,
            Credentials::new("APIKEY", "SECRET").with_passphrase("PASS"),
        )
        .with_clock(Box::new(move || now_ms));
        (kucoin, mock)
    }

    #[test]
    fn hedge_mode_is_refused_rather_than_ignored() {
        // KuCoin Futures is one-way only: there is no field on the order that
        // could carry a side. Placing it anyway would move the single net
        // position, which is what configuring hedge asked to avoid.
        let (hedged, _mock) = hedged_futures_client(1000);
        let err = hedged
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap_err();
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));

        // One-way on the same venue still places orders.
        let (one_way, _) = signed_futures_client(1000);
        let _ = one_way.place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)));
    }

    const KU_POSITIONS: &str = r#"{"code":"200000","data":[
        {"symbol":"XBTUSDTM","currentQty":3,"avgEntryPrice":20000.0,"markPrice":20100.0,"realLeverage":10.0,"unrealisedPnl":30.0,"crossMode":false}
    ]}"#;

    #[test]
    fn stp_maps_to_stp_flag() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"1"}}"#);
        kucoin
            .place_order(
                &OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
                    .with_stp(SelfTradePrevention::ExpireBoth),
            )
            .unwrap();
        let reqs = mock.recorded_requests();
        assert!(reqs[0].body.as_ref().unwrap().contains(r#""stp":"CB""#));
    }

    /// KuCoin stamps both public reads with `data.time`. Payloads captured from
    /// the live endpoints.
    #[test]
    fn the_public_reads_carry_the_venue_stamp() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"time":1788396666570,"last":"77083.8","buy":"77083.7","sell":"77083.8","vol":"1234.5"}}"#,
        );
        let ticker = kucoin.ticker(&symbol()).unwrap();
        assert_eq!(ticker.timestamp, 1_788_396_666_570);

        let (kucoin, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"time":1788396668372,"sequence":"36500054130","bids":[["77083.7","0.92"]],"asks":[["77083.8","0.06"]]}}"#,
        );
        let book = kucoin.order_book(&symbol(), 5).unwrap();
        assert_eq!(book.timestamp, 1_788_396_668_372);
    }

    #[test]
    fn place_batch_multi_per_order_results() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"data":[
            {"orderId":"o1","status":"success"},
            {"orderId":"","status":"fail","failMsg":"insufficient"}]}}"#,
        );
        let results = kucoin
            .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, "o1");
        assert!(matches!(
            results[1].as_ref().unwrap_err(),
            Error::OrderRejected { .. }
        ));
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/api/v1/orders/multi"));
    }

    #[test]
    fn cancel_batch_is_sequential() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{}}"#);
        mock.push_json(200, r#"{"code":"200000","data":{}}"#);
        kucoin
            .cancel_batch(&symbol(), &["1".to_string(), "2".to_string()])
            .unwrap();
        assert_eq!(mock.recorded_requests().len(), 2);
    }

    #[test]
    fn place_oco_is_a_single_order_list() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"oco1"}}"#);
        let legs = kucoin
            .place_oco(&OcoRequest::new(
                symbol(),
                OrderSide::Sell,
                dec!(1),
                dec!(110),
                dec!(95),
            ))
            .unwrap();
        assert_eq!(legs.len(), 1);
        assert_eq!(legs[0].id, "oco1");
        let reqs = mock.recorded_requests();
        assert!(reqs[0].url.contains("/api/v3/oco/order"));
        assert!(reqs[0]
            .body
            .as_ref()
            .unwrap()
            .contains(r#""stopPrice":"95""#));
    }

    #[test]
    fn amend_is_unsupported() {
        let (mut kucoin, _mock) = signed_client(1000);
        assert!(matches!(
            AdvancedOrders::amend_order(&mut kucoin, &symbol(), "1", Some(dec!(1)), None)
                .unwrap_err(),
            Error::Exchange { .. }
        ));
    }

    #[test]
    fn futures_client_uses_futures_host_and_contract_symbol() {
        let (kucoin, _mock) = signed_futures_client(1000);
        assert!(kucoin.rest_base.contains("api-futures.kucoin.com"));
        assert_eq!(KuCoin::futures_symbol(&symbol()), "XBTUSDTM");
    }

    #[test]
    fn futures_place_order_carries_leverage_and_contract_symbol() {
        let (mut kucoin, mock) = signed_futures_client(1000);
        kucoin.set_leverage(&symbol(), 5).unwrap();
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"9"}}"#);
        kucoin
            .place_order(&OrderRequest::market_buy(symbol(), dec!(2)))
            .unwrap();
        let reqs = mock.recorded_requests();
        let body = reqs[0].body.as_deref().unwrap();
        assert!(body.contains(r#""symbol":"XBTUSDTM""#));
        assert!(body.contains(r#""leverage":"5""#));
        assert!(reqs[0].url.contains("api-futures.kucoin.com"));
    }

    #[test]
    fn derivatives_positions_parse_contracts() {
        let (mut kucoin, mock) = signed_futures_client(1000);
        mock.push_json(200, KU_POSITIONS);
        let positions = Derivatives::positions(&mut kucoin, 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)); // contracts
        assert_eq!(positions[0].margin_mode, MarginMode::Isolated);
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/api/v1/positions"));
    }

    #[test]
    fn derivatives_set_margin_mode_changes_mode() {
        let (mut kucoin, mock) = signed_futures_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{}}"#);
        Derivatives::set_margin_mode(&mut kucoin, &symbol(), MarginMode::Cross).unwrap();
        let req = &mock.recorded_requests()[0];
        assert!(req.url.contains("/api/v1/position/changeMarginMode"));
        assert!(req
            .body
            .as_deref()
            .unwrap()
            .contains(r#""marginMode":"CROSS""#));
    }

    #[test]
    fn derivatives_close_position_reduce_only() {
        let (mut kucoin, mock) = signed_futures_client(1000);
        mock.push_json(200, KU_POSITIONS);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"9"}}"#);
        Derivatives::close_position(&mut kucoin, &symbol()).unwrap();
        let reqs = mock.recorded_requests();
        let body = reqs[1].body.as_deref().unwrap();
        assert!(body.contains(r#""side":"sell""#));
        assert!(body.contains(r#""reduceOnly":true"#));
    }

    #[test]
    fn ticker_from_stats() {
        let (kucoin, mock) = client();
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"last":"20000","buy":"19999","sell":"20001","vol":"1234"}}"#,
        );
        let ticker = kucoin.ticker(&symbol()).unwrap();
        assert_eq!(ticker.last, dec!(20000));
        assert_eq!(ticker.bid, dec!(19999));
        assert_eq!(ticker.ask, dec!(20001));
    }

    #[test]
    fn klines_field_order_and_reverse() {
        let (kucoin, mock) = client();
        // [time, open, close, high, low, volume, turnover], newest-first.
        mock.push_json(
            200,
            r#"{"code":"200000","data":[
            ["1700000060","105","105.5","106","104","2","0"],
            ["1700000000","100","105","110","95","12","0"]]}"#,
        );
        let candles = kucoin.klines(&symbol(), "1h", 2).unwrap();
        assert_eq!(candles[0].timestamp, 1_700_000_000);
        // open=100, high=110, low=95, close=105.
        assert!((candles[0].high - 110.0).abs() < 1e-9);
        assert!((candles[0].low - 95.0).abs() < 1e-9);
        assert!((candles[0].close - 105.0).abs() < 1e-9);
    }

    #[test]
    fn order_book_parses() {
        let (kucoin, mock) = client();
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"sequence":"55","bids":[["100","1"]],"asks":[["101","2"]]}}"#,
        );
        let book = kucoin.order_book(&symbol(), 20).unwrap();
        assert_eq!(book.last_update_id, 55);
        assert_eq!(book.bids[0], BookLevel::new(dec!(100), dec!(1)));
    }

    #[test]
    fn error_mapping() {
        let (kucoin, mock) = client();
        mock.push_json(200, r#"{"code":"200004","msg":"balance","data":null}"#);
        assert!(matches!(
            kucoin.ticker(&symbol()).unwrap_err(),
            Error::InsufficientBalance
        ));
    }

    #[test]
    fn place_order_signs_with_kc_headers_and_signed_passphrase() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"99"}}"#);
        let order = kucoin
            .place_order(
                &OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)).with_client_order_id("abc"),
            )
            .unwrap();
        assert_eq!(order.id, "99");
        assert_eq!(order.client_order_id.as_deref(), Some("abc"));

        let req = &mock.recorded_requests()[0];
        let header = |name: &str| {
            req.headers
                .iter()
                .find(|(k, _)| k == name)
                .map(|(_, v)| v.as_str())
                .unwrap()
        };
        let ts = header("KC-API-TIMESTAMP");
        assert_eq!(ts, "1000");
        let body = req.body.as_ref().unwrap();
        let prehash = format!("{ts}POST/api/v1/orders{body}");
        assert_eq!(
            header("KC-API-SIGN"),
            hmac_sha256_base64(b"SECRET", prehash.as_bytes())
        );
        // The passphrase header is itself HMAC-signed.
        assert_eq!(
            header("KC-API-PASSPHRASE"),
            hmac_sha256_base64(b"SECRET", b"PASS")
        );
        assert_eq!(header("KC-API-KEY-VERSION"), "2");
    }

    /// An unnamed order still gets a client id, and two of them get *different*
    /// ones.
    ///
    /// The generated id used to be the millisecond alone, so two orders placed
    /// inside the same millisecond carried the same id -- and KuCoin
    /// deduplicates on it, so the second was refused as a repeat of the first.
    /// The assertion is on that property rather than on an exact string: the
    /// value is seeded from the clock, and pinning it would only re-test the
    /// seed.
    #[test]
    fn place_order_generates_a_unique_client_oid_when_absent() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"1"}}"#);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"2"}}"#);

        let first = kucoin
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        let second = kucoin
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();

        let first_id = first.client_order_id.expect("an id was generated");
        let second_id = second.client_order_id.expect("an id was generated");
        // The id is not interpolated into the message. CodeQL traces any
        // value out of a credentials-holding client as sensitive, and the
        // rule is right in spirit even where this particular value is a
        // counter: a test message is still output.
        assert!(
            first_id.starts_with("wkex-"),
            "the generated id must carry the wkex- prefix"
        );
        assert_ne!(
            first_id, second_id,
            "two orders on a frozen clock must not share a client id"
        );
    }

    #[test]
    fn query_order_status_from_flags() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"200000","data":{"id":"99","clientOid":"","symbol":"BTC-USDT",
            "side":"sell","type":"limit","size":"2","dealSize":"0","price":"100",
            "isActive":false,"cancelExist":false}}"#,
        );
        let order = kucoin.query_order(&symbol(), "99").unwrap();
        // Inactive, not cancelled -> filled.
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.side, OrderSide::Sell);
    }

    #[test]
    fn balances_and_open_orders() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"200000","data":[{"currency":"USDT","available":"100.5","holds":"25.5"}]}"#,
        );
        let bals = kucoin.balances().unwrap();
        assert_eq!(bals[0].total(), dec!(126));

        mock.push_json(
            200,
            r#"{"code":"200000","data":{"items":[{"id":"7","clientOid":"","symbol":"ETH-USDT",
            "side":"buy","type":"limit","size":"1","dealSize":"0","price":"50",
            "isActive":true,"cancelExist":false}]}}"#,
        );
        let open = kucoin.open_orders(None).unwrap();
        assert_eq!(open[0].symbol, Symbol::new("ETH", "USDT"));
        assert_eq!(open[0].status, OrderStatus::New);
    }

    #[test]
    fn signed_requires_passphrase() {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let kucoin = KuCoin::with_credentials(
            Box::new(ArcTransport(mock)),
            &opts,
            Credentials::new("k", "s"),
        );
        assert!(matches!(
            kucoin.balances().unwrap_err(),
            Error::InvalidCredentials(_)
        ));
    }

    #[test]
    fn ws_parses_trade_ticker_book() {
        let ws = Arc::new(MockWsTransport::new());
        ws.push_connection(vec![
            Ok(Some(
                r#"{"type":"message","topic":"/market/match:BTC-USDT","subject":"trade.l3match",
                "data":{"symbol":"BTC-USDT","side":"buy","price":"100","size":"0.5","time":"1"}}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"type":"message","topic":"/market/level2:BTC-USDT","subject":"trade.l2update",
                "data":{"sequenceEnd":9,"changes":{"bids":[["100","1","9"]],"asks":[]}}}"#
                    .to_string(),
            )),
            Ok(Some(r#"{"type":"welcome"}"#.to_string())),
        ]);
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut kucoin = KuCoin::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        kucoin.subscribe_trades(&symbol()).unwrap();
        assert!(ws.sent()[0].contains(r#""topic":"/market/match:BTC-USDT""#));

        let events = kucoin.poll_events();
        assert_eq!(events.len(), 2);
        let Event::Trade(t) = &events[0] else {
            panic!("expected trade")
        };
        assert_eq!(t.aggressor, OrderSide::Buy);
        let Event::BookDelta(d) = &events[1] else {
            panic!("expected book delta")
        };
        assert_eq!(d.final_update_id, 9);
    }

    #[test]
    fn works_as_a_boxed_exchange() {
        let (kucoin, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"200000","data":{"orderId":"1"}}"#);
        let mut exchange: Box<dyn Exchange> = Box::new(kucoin);
        assert_eq!(exchange.name(), "kucoin");
        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("KuCoin {"));
        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 kucoin, http) = signed_client(1_000_000);

        // Shape verified against the live public endpoint on 2026-09-01.
        http.push_json(200, r#"{"code":"200000","data":1004500}"#);
        let offset = kucoin.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#"{"code":"200000","data":[]}"#);
        kucoin.balances().expect("balances must succeed");

        let requests = http.recorded_requests();
        let signed = requests.last().expect("a signed request was recorded");
        let header = signed
            .headers
            .iter()
            .find(|(name, _)| name == "KC-API-TIMESTAMP")
            .map(|(_, value)| value.as_str())
            .expect("signed request carries KC-API-TIMESTAMP");
        assert_eq!(header, "1004500", "the venue's time, not ours");
    }

    /// 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 (kucoin, mock) = client();
        mock.push_response(
            crate::transport::HttpResponse::new(
                200,
                r#"{"code":"429000","msg":"rate","data":null}"#,
            )
            .with_header("Retry-After", "2.5"),
        );
        let err = kucoin.ticker(&symbol()).unwrap_err();
        let wait = std::time::Duration::from_millis(2500);
        assert!(matches!(err, Error::RateLimited { retry_after: Some(d) } if d == wait));
    }
}