wickra-exchange-core 0.2.0

//! Bitget (v2 API) — the fourth exchange.
//!
//! Bitget's signing is close to OKX's — base64(HMAC-SHA256) with a passphrase —
//! but over a **millisecond** timestamp rather than ISO-8601, with `ACCESS-*`
//! headers, concatenated symbols (`BTCUSDT`) and a `{code, msg, data}` envelope
//! whose success code is the string `"00000"`.
//!
//! On a futures [`MarketType`](crate::MarketType), `query_order`/`cancel_order`/
//! `open_orders` route to the mix order endpoints (`/api/v2/mix/order/detail`,
//! `/cancel-order`, `/orders-pending`) with `productType=USDT-FUTURES` and the
//! mix order object (`state` instead of `status`, `entrustedList` wrapper).
//!
//! [`AdvancedOrders`]: STP via `stpMode` on order create and native batch
//! place/cancel — spot (`/api/v2/spot/trade/batch-orders`, `.../batch-cancel-order`)
//! or mix (`/api/v2/mix/order/batch-place-order`, `.../batch-cancel-orders`),
//! per-order results. Bitget has no in-place amend and no OCO order-list — both
//! documented gaps.

use crate::clock::ServerClock;
use crate::credentials::Credentials;
use crate::error::{Error, Result};
use crate::events::{BookDelta, BookLevel, Event, OrderBookSnapshot, TradePrint};
use crate::feeds::{
    DerivativesChannel, DerivativesFeed, FundingRate, 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::Decimal;
use serde::Deserialize;
use std::collections::HashMap;
use std::fmt;
use wickra_core::Candle;

/// The current Unix time in milliseconds, from the system clock.
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
}

/// Bitget's `tradeSide` for a hedged futures order: `open` grows the side the
/// order's own side implies, `close` reduces the opposite one. One-way orders
/// carry `reduceOnly` instead and no `tradeSide` at all.
fn trade_side(request: &OrderRequest) -> &'static str {
    if request.reduce_only {
        "close"
    } else {
        "open"
    }
}

/// Bitget's wire word for a margin mode. Cross is `crossed`, not `cross`.
fn margin_wire(mode: MarginMode) -> &'static str {
    match mode {
        MarginMode::Cross => "crossed",
        MarginMode::Isolated => "isolated",
    }
}

/// A Bitget client over injected transports.
pub struct Bitget {
    http: Box<dyn HttpTransport>,
    ws: Option<Box<dyn WsTransport>>,
    rest_base: String,
    market_type: MarketType,
    /// The margin mode every futures order carries, kept in step by
    /// [`set_margin_mode`](Self::set_margin_mode).
    margin_mode: MarginMode,
    /// One-way or hedge. Bitget's v2 mix API says which side a hedged order
    /// acts on with `tradeSide`, alongside the plain buy/sell `side`.
    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 previous fallback was the order's index within its batch, which
    /// is unique inside one batch and repeats across every batch after it.
    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 wire symbol.
    ///
    /// Bitget publishes funding and mark/index inside the ordinary `ticker`
    /// frame, so one frame can answer three subscriptions. Only the prints
    /// that were asked for are emitted: a caller watching prices does not
    /// start receiving funding it never subscribed to.
    derivatives_channels: Vec<(String, DerivativesChannel)>,
    /// 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 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 Bitget {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Bitget")
            .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("private_connection", &self.private_connection.is_some())
            .field("user_data_active", &self.user_data_active)
            .finish_non_exhaustive()
    }
}

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

    fn build(
        http: Box<dyn HttpTransport>,
        options: &ExchangeOptions,
        credentials: Option<Credentials>,
    ) -> Self {
        Self {
            http,
            ws: None,
            rest_base: "https://api.bitget.com".to_string(),
            market_type: options.market_type,
            margin_mode: options.margin_mode,
            position_mode: options.position_mode,
            credentials,
            now_ms: Box::new(system_now_ms),
            client_ids: ClientIdGenerator::with_seed(
                "wbatch",
                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(),
            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/v2/public/time`.
    ///
    /// Explicit rather than automatic: it costs a request, and a client should
    /// not make one the caller did not ask for. Call it once after connecting,
    /// and again if the process runs long enough for drift to matter.
    ///
    /// Returns the new offset in milliseconds (`server - local`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the response cannot be parsed.
    pub fn sync_time(&mut self) -> Result<i64> {
        let local_before = (self.now_ms)();
        let value = self.get("/api/v2/public/time", "")?;
        let server_ms: i64 = value
            .get("serverTime")
            .and_then(serde_json::Value::as_str)
            .ok_or_else(|| Error::Deserialization("public/time: no serverTime".into()))?
            .parse()
            .map_err(|_| Error::Deserialization("public/time: serverTime not an integer".into()))?;
        self.clock.sync(local_before, server_ms);
        Ok(self.clock.offset_ms())
    }

    /// Whether this client targets a USDⓈ-M futures market (Bitget's `mix`
    /// product line, `USDT-FUTURES` / margin coin `USDT`) rather than spot.
    fn is_futures(&self) -> bool {
        self.market_type.is_derivatives()
    }

    /// The `instType` every WebSocket subscription is keyed by.
    ///
    /// Bitget v2 serves one socket URL for every product and distinguishes them
    /// here, exactly as the REST paths distinguish them with `productType`. Both
    /// subscribe paths hardcoded `SPOT`, so a futures client streamed the spot
    /// book, the spot trades and -- through the private stream -- the spot
    /// account, while its REST calls read futures. The two now agree.
    fn inst_type(&self) -> &'static str {
        if self.is_futures() {
            "USDT-FUTURES"
        } else {
            "SPOT"
        }
    }

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

    /// Build an authenticated Bitget 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 Bitget wire symbol for a canonical [`Symbol`] (`BTC/USDT` -> `BTCUSDT`).
    #[must_use]
    pub fn wire_symbol(symbol: &Symbol) -> String {
        symbol.to_concatenated()
    }

    /// A ticker for `symbol`.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the symbol is unknown.
    pub fn ticker(&self, symbol: &Symbol) -> Result<Ticker> {
        let (path, query) = if self.is_futures() {
            (
                "/api/v2/mix/market/ticker",
                format!(
                    "symbol={}&productType=USDT-FUTURES",
                    Self::wire_symbol(symbol)
                ),
            )
        } else {
            (
                "/api/v2/spot/market/tickers",
                format!("symbol={}", Self::wire_symbol(symbol)),
            )
        };
        let data = self.get(path, &query)?;
        let list: Vec<RawTicker> = parse_json(data)?;
        let entry = list
            .into_iter()
            .next()
            .ok_or_else(|| Error::NotFound(format!("no ticker for {symbol}")))?;
        Ok(Ticker {
            symbol: symbol.clone(),
            last: parse_decimal(&entry.last_pr)?,
            bid: parse_decimal(&entry.bid_pr)?,
            ask: parse_decimal(&entry.ask_pr)?,
            volume: parse_decimal(&entry.base_volume)?,
            timestamp: entry.ts.parse().unwrap_or(0),
        })
    }

    /// Up to `limit` candles for `symbol` at `interval` (unified). Bitget returns
    /// oldest-first, which is already chronological.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or a row cannot be parsed.
    pub fn klines(&self, symbol: &Symbol, interval: &str, limit: u32) -> Result<Vec<Candle>> {
        let (path, extra) = if self.is_futures() {
            ("/api/v2/mix/market/candles", "&productType=USDT-FUTURES")
        } else {
            ("/api/v2/spot/market/candles", "")
        };
        let query = format!(
            "symbol={}&granularity={}&limit={limit}{extra}",
            Self::wire_symbol(symbol),
            map_granularity(interval),
        );
        let data = self.get(path, &query)?;
        let rows: Vec<Vec<String>> = parse_json(data)?;
        rows.iter().map(|row| parse_kline_row(row)).collect()
    }

    /// A depth snapshot of `symbol` up to `depth` levels per side.
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails or the response cannot be parsed.
    pub fn order_book(&self, symbol: &Symbol, depth: u32) -> Result<OrderBookSnapshot> {
        let (path, query) = if self.is_futures() {
            (
                "/api/v2/mix/market/merge-depth",
                format!(
                    "symbol={}&productType=USDT-FUTURES&limit={depth}",
                    Self::wire_symbol(symbol)
                ),
            )
        } else {
            (
                "/api/v2/spot/market/orderbook",
                format!("symbol={}&limit={depth}", Self::wire_symbol(symbol)),
            )
        };
        let data = self.get(path, &query)?;
        let raw: RawDepth = parse_json(data)?;
        Ok(OrderBookSnapshot {
            symbol: symbol.clone(),
            last_update_id: raw.ts.parse().unwrap_or(0),
            bids: parse_levels(&raw.bids)?,
            asks: parse_levels(&raw.asks)?,
            timestamp: raw.ts.parse().unwrap_or(0),
        })
    }

    /// Subscribe to a pushed derivatives channel.
    ///
    /// Bitget publishes the funding rate, the mark price and the index price
    /// inside the mix `ticker` frame -- all three in one reading, taken at one
    /// moment -- so `Funding` and `MarkIndex` are the same subscription seen
    /// two ways.
    ///
    /// `Liquidations` is refused: Bitget publishes no public stream of forced
    /// orders. Subscribing to nothing and waiting would be the worse answer.
    ///
    /// # 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(
                "Bitget",
                "a derivatives channel on a spot client",
                "funding and mark/index exist only on the futures venue",
            ));
        }
        if channel == DerivativesChannel::Liquidations {
            return Err(Error::unsupported_field(
                "Bitget",
                "a liquidations channel",
                "Bitget publishes no public stream of forced orders",
            ));
        }
        let wire = Self::wire_symbol(symbol);
        if !self
            .derivatives_channels
            .iter()
            .any(|(w, c)| w == &wire && *c == channel)
        {
            self.derivatives_channels.push((wire, channel));
        }
        self.subscribe(symbol, "ticker")
    }

    /// The current open interest (`GET /api/v2/mix/market/open-interest`).
    ///
    /// Polled, not pushed. The figure is in base currency.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, if the request fails, or if the
    /// venue returns no data point.
    pub fn open_interest(&self, symbol: &Symbol) -> Result<OpenInterest> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "Bitget",
                "open interest on a spot client",
                "open interest is a futures figure",
            ));
        }
        let query = format!(
            "symbol={}&productType=USDT-FUTURES",
            Self::wire_symbol(symbol)
        );
        let data = self.get("/api/v2/mix/market/open-interest", &query)?;
        let point = data
            .get("openInterestList")
            .and_then(serde_json::Value::as_array)
            .and_then(|rows| rows.first())
            .ok_or_else(|| Error::NotFound("no open-interest data point".to_string()))?;
        Ok(OpenInterest {
            symbol: symbol.clone(),
            open_interest: parse_decimal(field_str(point, "size")?)?,
            timestamp: opt_str(&data, "ts").parse().unwrap_or(0),
        })
    }

    /// The current long/short account ratio
    /// (`GET /api/v2/mix/market/account-long-short`), most recent point.
    ///
    /// Polled, not pushed. Bitget publishes the two proportions directly, as
    /// Binance does, so nothing is derived here.
    ///
    /// # Errors
    /// Returns an [`Error`] on a spot client, if the request fails, or if the
    /// venue returns no data point.
    pub fn long_short_ratio(&self, symbol: &Symbol) -> Result<LongShortRatio> {
        if !self.is_futures() {
            return Err(Error::unsupported_field(
                "Bitget",
                "long/short positioning on a spot client",
                "positioning is a futures figure",
            ));
        }
        let query = format!(
            "symbol={}&productType=USDT-FUTURES&period=5m",
            Self::wire_symbol(symbol)
        );
        let data = self.get("/api/v2/mix/market/account-long-short", &query)?;
        let point = data
            .as_array()
            .and_then(|rows| rows.first())
            .ok_or_else(|| Error::NotFound("no long/short data point".to_string()))?;
        Ok(LongShortRatio {
            symbol: symbol.clone(),
            long_size: parse_decimal(field_str(point, "longAccountRatio")?)?,
            short_size: parse_decimal(field_str(point, "shortAccountRatio")?)?,
            timestamp: opt_str(point, "ts").parse().unwrap_or(0),
        })
    }

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

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

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

    fn subscribe(&mut self, symbol: &Symbol, channel: &str) -> Result<()> {
        self.ensure_market_is_routed()?;
        let wire = Self::wire_symbol(symbol);
        if self.connection.is_none() {
            let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
            let connection = ws.connect("wss://ws.bitget.com/v2/ws/public")?;
            self.connection = Some(connection);
        }
        let message = format!(
            r#"{{"op":"subscribe","args":[{{"instType":"{}","channel":"{channel}","instId":"{wire}"}}]}}"#,
            self.inst_type()
        );
        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(|| Symbol::new(wire, ""))
        };
        let channels = self.derivatives_channels.clone();
        let subscribed = |wire: &str, channel: DerivativesChannel| {
            channels.iter().any(|(w, c)| w == wire && *c == channel)
        };
        let mut events = Vec::new();
        if let Some(connection) = self.connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Ok(mut parsed) = parse_ws_message(&frame, &resolve, &subscribed) {
                    events.append(&mut parsed);
                }
            }
        }
        // Drain the private user-data stream (orders/account channels), if open.
        if let Some(connection) = self.private_connection.as_mut() {
            while let Ok(Some(frame)) = connection.recv() {
                if let Ok(mut parsed) = parse_ws_message(&frame, &resolve, &subscribed) {
                    events.append(&mut parsed);
                }
            }
        }
        // A dropped private stream is re-subscribed with a fresh op:login (the
        // signature is time-bound, so a stale replay would be rejected).
        if self.user_data_active
            && self
                .private_connection
                .as_ref()
                .is_some_and(|c| !c.is_connected())
        {
            events.push(Event::Disconnected);
            self.private_connection = None;
            if self.subscribe_user_data().is_ok() {
                events.push(Event::Reconnected);
            }
        }
        let url = "wss://ws.bitget.com/v2/ws/public";
        crate::wsutil::reconnect_if_dropped(
            self.ws.as_deref(),
            url,
            &mut self.connection,
            &self.sub_messages,
            &mut events,
        );
        events
    }

    /// Open the private user-data stream (`wss://ws.bitget.com/v2/ws/private`).
    /// Logs in with an `op:login` frame (sign = base64(HMAC-SHA256) over
    /// `<epochSeconds>GET/user/verify`), then subscribes to the `orders` and
    /// `account` channels. Afterwards [`poll_events`](Self::poll_events) also
    /// surfaces the account's own [`Event::OrderUpdate`] and [`Event::BalanceUpdate`].
    ///
    /// A dropped private stream is re-subscribed automatically on the next
    /// [`poll_events`](Self::poll_events); call
    /// [`keepalive_user_data`](Self::keepalive_user_data) periodically to keep it
    /// from being dropped for inactivity.
    ///
    /// # Errors
    /// Returns [`Error::InvalidCredentials`] without credentials or a passphrase,
    /// [`Error::NotConnected`] without a WebSocket transport, or another
    /// [`Error`] if the request fails.
    pub fn subscribe_user_data(&mut self) -> Result<()> {
        self.ensure_market_is_routed()?;
        let creds = self.credentials.as_ref().ok_or(Error::InvalidCredentials(
            "user-data stream requires credentials",
        ))?;
        let passphrase = creds
            .passphrase
            .as_deref()
            .ok_or(Error::InvalidCredentials("Bitget requires a passphrase"))?;
        // Bitget WS login signs `<timestamp>GET/user/verify`, timestamp in seconds.
        let timestamp = self.signed_now_ms() / 1000;
        let sign = hmac_sha256_base64(
            creds.api_secret.as_bytes(),
            format!("{timestamp}GET/user/verify").as_bytes(),
        );
        let login = format!(
            r#"{{"op":"login","args":[{{"apiKey":"{}","passphrase":"{passphrase}","timestamp":"{timestamp}","sign":"{sign}"}}]}}"#,
            creds.api_key
        );
        let inst_type = self.inst_type();
        let subscribe = format!(
            r#"{{"op":"subscribe","args":[{{"instType":"{inst_type}","channel":"orders","instId":"default"}},{{"instType":"{inst_type}","channel":"account","coin":"default"}}]}}"#
        );
        let ws = self.ws.as_ref().ok_or(Error::NotConnected)?;
        let mut connection = ws.connect("wss://ws.bitget.com/v2/ws/private")?;
        connection.send(&login)?;
        connection.send(&subscribe)?;
        self.private_connection = Some(connection);
        self.user_data_active = true;
        Ok(())
    }

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

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

    /// Place a trigger order through Bitget's plan-order endpoint.
    ///
    /// Bitget calls a conditional order a *plan* order and serves it from its
    /// own path -- `/api/v2/spot/trade/place-plan-order` or the mix equivalent
    /// -- so the trigger is not a field on the ordinary call.
    ///
    /// `triggerType` names **which price arms the stop**, and Bitget will not
    /// choose: `fill_price` watches the last trade, `mark_price` the mark. On a
    /// futures position the mark is what liquidates it, so that is what a stop
    /// there is meant to watch; on spot there is no mark, so the traded price is
    /// the only honest answer. Leaving it out would take whichever default the
    /// venue happens to hold.
    ///
    /// # Errors
    /// Returns an [`Error`] if the trigger price is missing or the request
    /// fails.
    fn place_plan_order(&self, request: &OrderRequest) -> Result<Order> {
        let stop = request
            .stop_price
            .ok_or(Error::InvalidOrder("a trigger order requires a stop price"))?;
        let futures = self.is_futures();
        let mut body = serde_json::json!({
            "symbol": Self::wire_symbol(&request.symbol),
            "side": side_str(request.side),
            "orderType": order_type_str(request.order_type),
            "size": format_decimal(request.quantity),
            "triggerPrice": format_decimal(stop),
            // The price that arms the stop: the mark on futures, because that
            // is what a position is liquidated against; the traded price on
            // spot, where no mark exists.
            "triggerType": if futures { "mark_price" } else { "fill_price" },
        });
        if let Some(price) = request.price {
            body["executePrice"] = serde_json::json!(format_decimal(price));
        }
        if let Some(id) = &request.client_order_id {
            body["clientOid"] = serde_json::json!(id.clone());
        }
        let path = if futures {
            body["productType"] = serde_json::json!("USDT-FUTURES");
            body["marginCoin"] = serde_json::json!("USDT");
            body["marginMode"] = serde_json::json!(margin_wire(self.margin_mode));
            // `normal_plan` is the ordinary trigger. The other plan types are
            // the position-attached take-profit and stop-loss, which belong to
            // an open position rather than to this request.
            body["planType"] = serde_json::json!("normal_plan");
            if self.position_mode == PositionMode::Hedge {
                body["tradeSide"] = serde_json::json!(trade_side(request));
            } else if request.reduce_only {
                body["reduceOnly"] = serde_json::json!("YES");
            }
            "/api/v2/mix/order/place-plan-order"
        } else {
            "/api/v2/spot/trade/place-plan-order"
        };
        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: request.client_order_id.clone(),
            symbol: request.symbol.clone(),
            side: request.side,
            order_type: request.order_type,
            status: OrderStatus::New,
            quantity: request.quantity,
            filled_quantity: Decimal::ZERO,
            price: request.price,
            average_price: None,
        })
    }

    pub fn place_order(&self, request: &OrderRequest) -> Result<Order> {
        self.ensure_reduce_only_is_reducible(request)?;
        request.validate()?;
        // A trigger is a different endpoint here, not a field on this one.
        if request.order_type.is_trigger() {
            return self.place_plan_order(request);
        }
        let force = force_for(request)?;
        let mut body = serde_json::json!({
            "symbol": Self::wire_symbol(&request.symbol),
            "side": side_str(request.side),
            "orderType": order_type_str(request.order_type),
            "force": force,
            "size": format_decimal(request.quantity),
        });
        if let Some(price) = request.price {
            body["price"] = serde_json::json!(format_decimal(price));
        }
        if let Some(id) = &request.client_order_id {
            body["clientOid"] = serde_json::json!(id.clone());
        }
        if let Some(mode) = stp_mode_str(request.stp) {
            body["stpMode"] = serde_json::json!(mode);
        }
        let path = if self.is_futures() {
            body["productType"] = serde_json::json!("USDT-FUTURES");
            body["marginMode"] = serde_json::json!(margin_wire(self.margin_mode));
            body["marginCoin"] = serde_json::json!("USDT");
            if self.position_mode == PositionMode::Hedge {
                // A hedged account holds both sides at once, so `tradeSide`
                // says whether this order grows a side or reduces one.
                // `reduceOnly` is the one-way spelling of the same thing and
                // Bitget does not accept the two together.
                body["tradeSide"] = serde_json::json!(trade_side(request));
            } else if request.reduce_only {
                body["reduceOnly"] = serde_json::json!("YES");
            }
            "/api/v2/mix/order/place-order"
        } else {
            "/api/v2/spot/trade/place-order"
        };
        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: (!placed.client_oid.is_empty()).then_some(placed.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<()> {
        if self.is_futures() {
            let body = serde_json::json!({
                "symbol": Self::wire_symbol(symbol),
                "productType": "USDT-FUTURES",
                "marginCoin": "USDT",
                "orderId": order_id,
            });
            self.signed_request(
                HttpMethod::Post,
                "/api/v2/mix/order/cancel-order",
                "",
                &body.to_string(),
            )?;
            return Ok(());
        }
        let body = serde_json::json!({
            "symbol": Self::wire_symbol(symbol),
            "orderId": order_id,
        });
        self.signed_request(
            HttpMethod::Post,
            "/api/v2/spot/trade/cancel-order",
            "",
            &body.to_string(),
        )?;
        Ok(())
    }

    /// Query a single order by venue id.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the order is unknown.
    pub fn query_order(&self, symbol: &Symbol, order_id: &str) -> Result<Order> {
        if self.is_futures() {
            let query = format!(
                "symbol={}&productType=USDT-FUTURES&orderId={order_id}",
                Self::wire_symbol(symbol)
            );
            // The mix order-detail endpoint returns a single order object.
            let data =
                self.signed_request(HttpMethod::Get, "/api/v2/mix/order/detail", &query, "")?;
            let raw: RawOrder = parse_json(data)?;
            return order_from_raw(symbol.clone(), &raw);
        }
        let query = format!("orderId={order_id}");
        let data =
            self.signed_request(HttpMethod::Get, "/api/v2/spot/trade/orderInfo", &query, "")?;
        let list: Vec<RawOrder> = parse_json(data)?;
        let raw = list
            .into_iter()
            .next()
            .ok_or_else(|| Error::NotFound(format!("order {order_id}")))?;
        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>> {
        if self.is_futures() {
            let mut query = "productType=USDT-FUTURES".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/v2/mix/order/orders-pending",
                &query,
                "",
            )?;
            // The mix pending endpoint wraps the orders in an `entrustedList`.
            let pending: RawMixPending = parse_json(data)?;
            return pending
                .entrusted_list
                .iter()
                .map(|raw| {
                    let sym = symbol
                        .cloned()
                        .unwrap_or_else(|| split_wire_symbol(&raw.symbol));
                    order_from_raw(sym, raw)
                })
                .collect();
        }
        let query = match symbol {
            Some(s) => format!("symbol={}", Self::wire_symbol(s)),
            None => String::new(),
        };
        let data = self.signed_request(
            HttpMethod::Get,
            "/api/v2/spot/trade/unfilled-orders",
            &query,
            "",
        )?;
        let list: Vec<RawOrder> = parse_json(data)?;
        list.iter()
            .map(|raw| {
                let sym = symbol
                    .cloned()
                    .unwrap_or_else(|| split_wire_symbol(&raw.symbol));
                order_from_raw(sym, raw)
            })
            .collect()
    }

    /// Spot account balances.
    ///
    /// # Errors
    /// Returns an [`Error`] if credentials are missing or the request fails.
    pub fn balances(&self) -> Result<Vec<Balance>> {
        if self.is_futures() {
            let data = self.signed_request(
                HttpMethod::Get,
                "/api/v2/mix/account/accounts",
                "productType=USDT-FUTURES",
                "",
            )?;
            let list: Vec<RawMixAccount> = parse_json(data)?;
            return Ok(list
                .iter()
                .map(|a| Balance {
                    asset: a.margin_coin.clone(),
                    free: dec_or_zero(&a.available),
                    locked: dec_or_zero(&a.locked),
                })
                .collect());
        }
        let data = self.signed_request(HttpMethod::Get, "/api/v2/spot/account/assets", "", "")?;
        let list: Vec<RawAsset> = parse_json(data)?;
        Ok(list
            .iter()
            .map(|a| Balance {
                asset: a.coin.clone(),
                free: dec_or_zero(&a.available),
                locked: dec_or_zero(&a.frozen),
            })
            .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| {
            super::attribute_to_http_status(&response, e).with_retry_after(response.retry_after())
        })
    }

    /// The client order id to send for `request`: the caller's, or a generated
    /// one.
    ///
    /// The generated half used to be the order's index within its batch, so the
    /// first order of every batch was `wbatch-0`. Bitget deduplicates on this
    /// id, so the second batch's first order looked like a repeat of the first
    /// batch's.
    fn batch_client_oid(&self, request: &OrderRequest) -> String {
        request
            .client_order_id
            .clone()
            .unwrap_or_else(|| self.client_ids.next_id())
    }

    /// Sign with the `ACCESS-*` headers: base64(HMAC-SHA256) over
    /// `msTimestamp + METHOD + requestPath + body`.
    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("Bitget requires a passphrase"))?;
        let timestamp = self.signed_now_ms().to_string();
        let request_path = if query.is_empty() {
            path.to_string()
        } else {
            format!("{path}?{query}")
        };
        let prehash = format!("{timestamp}{}{request_path}{body}", method.as_str());
        let signature = hmac_sha256_base64(creds.api_secret.as_bytes(), prehash.as_bytes());
        let url = format!("{}{request_path}", self.rest_base);
        let mut request = HttpRequest::new(method, url)
            .with_header("ACCESS-KEY", creds.api_key.clone())
            .with_header("ACCESS-SIGN", signature)
            .with_header("ACCESS-TIMESTAMP", timestamp)
            .with_header("ACCESS-PASSPHRASE", passphrase.to_string())
            .with_header("locale", "en-US");
        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| {
            super::attribute_to_http_status(&response, e).with_retry_after(response.retry_after())
        })
    }
}

/// Map a unified interval to Bitget's `granularity` (`1min`, `1h`, `1day`).
fn map_granularity(interval: &str) -> String {
    match interval {
        "1m" => "1min",
        "3m" => "3min",
        "5m" => "5min",
        "15m" => "15min",
        "30m" => "30min",
        "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 != "00000" {
        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 {
        "429" | "30007" => Error::RateLimited { retry_after: None },
        "40009" | "40012" | "40037" => Error::Auth(msg.to_string()),
        "43012" | "43011" => Error::InsufficientBalance,
        "40034" | "43001" => 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 Bitget `force` value for a request. `post_only` is one of the values that
/// field takes, alongside `gtc`/`ioc`/`fok`, so setting both asks for two things
/// in one slot: that is refused rather than resolved by dropping the
/// time-in-force, which is what this builder used to do silently.
fn force_for(request: &OrderRequest) -> Result<&'static str> {
    match (request.post_only, request.time_in_force) {
        (true, TimeInForce::Gtc) => Ok("post_only"),
        (true, _) => Err(Error::unsupported_field(
            "Bitget",
            "post_only together with a non-GTC time-in-force",
            "`post_only` is one of the values Bitget's `force` field takes",
        )),
        (false, tif) => Ok(force_str(tif)),
    }
}

fn force_str(tif: TimeInForce) -> &'static str {
    match tif {
        TimeInForce::Gtc => "gtc",
        TimeInForce::Ioc => "ioc",
        TimeInForce::Fok => "fok",
    }
}

/// The Bitget `stpMode` value for a self-trade-prevention policy, or `None` to omit.
fn stp_mode_str(stp: SelfTradePrevention) -> Option<&'static str> {
    match stp {
        SelfTradePrevention::None => None,
        SelfTradePrevention::ExpireMaker => Some("cancel_maker"),
        SelfTradePrevention::ExpireTaker => Some("cancel_taker"),
        SelfTradePrevention::ExpireBoth => Some("cancel_both"),
    }
}

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 parse_status(raw: &str) -> Result<OrderStatus> {
    match raw {
        "live" | "new" | "init" => Ok(OrderStatus::New),
        "partially_filled" => Ok(OrderStatus::PartiallyFilled),
        "filled" | "full_fill" => Ok(OrderStatus::Filled),
        "cancelled" | "canceled" => Ok(OrderStatus::Canceled),
        other => Err(Error::Deserialization(format!("unknown status {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)
}

const KNOWN_QUOTES: &[&str] = &["USDT", "USDC", "EUR", "BTC", "ETH", "USD"];

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

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> {
    // Bitget candle: [ts, open, high, low, close, baseVol, quoteVol].
    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 ts 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}")))
    };
    Candle::new(f(1)?, f(2)?, f(3)?, f(4)?, f(5)?, ts)
        .map_err(|e| Error::Deserialization(e.to_string()))
}

fn order_from_raw(symbol: Symbol, raw: &RawOrder) -> Result<Order> {
    Ok(Order {
        id: raw.order_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: parse_status(&raw.status)?,
        quantity: parse_decimal(&raw.size)?,
        filled_quantity: dec_or_zero(&raw.base_volume),
        price: nonzero_decimal(&raw.price),
        average_price: nonzero_decimal(&raw.price_avg),
    })
}

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

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

fn parse_ws_message(
    text: &str,
    resolve: &impl Fn(&str) -> Symbol,
    subscribed: &impl Fn(&str, DerivativesChannel) -> bool,
) -> Result<Vec<Event>> {
    let value: serde_json::Value =
        serde_json::from_str(text).map_err(|e| Error::Deserialization(e.to_string()))?;
    let arg = value.get("arg");
    let Some(channel) = arg
        .and_then(|a| a.get("channel"))
        .and_then(serde_json::Value::as_str)
    else {
        return Ok(Vec::new());
    };
    let wire = arg
        .and_then(|a| a.get("instId"))
        .and_then(serde_json::Value::as_str)
        .unwrap_or("");
    let symbol = resolve(wire);
    let empty = Vec::new();
    let data = value
        .get("data")
        .and_then(serde_json::Value::as_array)
        .unwrap_or(&empty);

    if channel == "trade" {
        data.iter()
            .map(|t| {
                Ok(Event::Trade(TradePrint {
                    symbol: symbol.clone(),
                    price: parse_decimal(field_str(t, "price")?)?,
                    quantity: parse_decimal(field_str(t, "size")?)?,
                    aggressor: parse_side(field_str(t, "side")?)?,
                    timestamp: opt_str(t, "ts").parse().unwrap_or(0),
                }))
            })
            .collect()
    } else if channel == "ticker" {
        // One frame, up to three answers. The mix ticker carries the funding
        // rate, the mark price and the index price beside the quote, all read at
        // the same moment -- so a `MarkIndex` from here is one observation
        // rather than two stitched together. Only what was subscribed to is
        // emitted, so a caller watching prices is not handed funding prints.
        let mut out = Vec::new();
        for t in data {
            let wire = opt_str(t, "instId");
            let timestamp = opt_str(t, "ts").parse().unwrap_or(0);
            out.push(Event::Ticker(Ticker {
                symbol: symbol.clone(),
                last: parse_decimal(field_str(t, "lastPr")?)?,
                bid: dec_or_zero(opt_str(t, "bidPr")),
                ask: dec_or_zero(opt_str(t, "askPr")),
                volume: dec_or_zero(opt_str(t, "baseVolume")),
                timestamp,
            }));
            let mark = parse_decimal(opt_str(t, "markPrice")).ok();
            if subscribed(wire, DerivativesChannel::Funding) {
                if let (Some(rate), Some(mark_price)) =
                    (parse_decimal(opt_str(t, "fundingRate")).ok(), mark)
                {
                    out.push(Event::Derivatives(DerivativesFeed::Funding(FundingRate {
                        symbol: symbol.clone(),
                        rate,
                        mark_price,
                        timestamp,
                    })));
                }
            }
            if subscribed(wire, DerivativesChannel::MarkIndex) {
                if let (Some(mark_price), Some(index_price)) =
                    (mark, parse_decimal(opt_str(t, "indexPrice")).ok())
                {
                    out.push(Event::Derivatives(DerivativesFeed::MarkIndex(MarkIndex {
                        symbol: symbol.clone(),
                        mark_price,
                        index_price,
                        timestamp,
                    })));
                }
            }
        }
        Ok(out)
    } else if channel == "books" {
        let action = value.get("action").and_then(serde_json::Value::as_str);
        data.iter()
            .map(|b| {
                let update_id = opt_str(b, "ts").parse().unwrap_or(0);
                let bids = parse_ws_levels(b.get("bids"))?;
                let asks = parse_ws_levels(b.get("asks"))?;
                if action == Some("snapshot") {
                    Ok(Event::BookSnapshot(OrderBookSnapshot {
                        symbol: symbol.clone(),
                        last_update_id: update_id,
                        bids,
                        asks,
                        timestamp: i64::try_from(update_id).unwrap_or(0),
                    }))
                } else {
                    Ok(Event::BookDelta(BookDelta {
                        symbol: symbol.clone(),
                        first_update_id: update_id,
                        final_update_id: update_id,
                        bids,
                        asks,
                        timestamp: i64::try_from(update_id).unwrap_or(0),
                    }))
                }
            })
            .collect()
    } else if channel == "orders" {
        // Private order channel: each element carries its own `instId`, which
        // `RawOrder` picks up as the wire symbol via the `instId` alias.
        data.iter()
            .map(|raw| {
                let order: RawOrder = serde_json::from_value(raw.clone())
                    .map_err(|e| Error::Deserialization(e.to_string()))?;
                let order_symbol = split_wire_symbol(&order.symbol);
                Ok(Event::OrderUpdate(order_from_raw(order_symbol, &order)?))
            })
            .collect()
    } else if channel == "account" {
        // Private account channel: `data` is an array of per-coin balances,
        // emitted together as one balance-update snapshot.
        let balances = data
            .iter()
            .map(|raw| {
                let asset: RawAsset = serde_json::from_value(raw.clone())
                    .map_err(|e| Error::Deserialization(e.to_string()))?;
                Ok(Balance {
                    asset: asset.coin,
                    free: dec_or_zero(&asset.available),
                    locked: dec_or_zero(&asset.frozen),
                })
            })
            .collect::<Result<Vec<_>>>()?;
        Ok(vec![Event::BalanceUpdate(balances)])
    } else {
        Ok(Vec::new())
    }
}

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

#[derive(Deserialize)]
struct RawTicker {
    #[serde(rename = "lastPr")]
    last_pr: String,
    #[serde(rename = "bidPr")]
    bid_pr: String,
    #[serde(rename = "askPr")]
    ask_pr: String,
    #[serde(rename = "baseVolume")]
    base_volume: String,
    /// Bitget stamps every ticker; verified against the live endpoint.
    #[serde(default)]
    ts: String,
}

#[derive(Deserialize)]
struct RawDepth {
    bids: Vec<[String; 2]>,
    asks: Vec<[String; 2]>,
    #[serde(default)]
    ts: String,
}

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

#[derive(Deserialize)]
struct RawOrder {
    // REST reports `symbol`; the private `orders` stream reports `instId`.
    #[serde(default, alias = "instId")]
    symbol: String,
    #[serde(rename = "orderId")]
    order_id: String,
    #[serde(rename = "clientOid", default)]
    client_oid: String,
    side: String,
    #[serde(rename = "orderType")]
    order_type: String,
    // Spot reports `status`; the mix (futures) order object reports `state`.
    #[serde(alias = "state")]
    status: String,
    size: String,
    #[serde(rename = "baseVolume", default)]
    base_volume: String,
    #[serde(default)]
    price: String,
    #[serde(rename = "priceAvg", default)]
    price_avg: String,
}

#[derive(Deserialize)]
struct RawAsset {
    coin: String,
    #[serde(default)]
    available: String,
    #[serde(default)]
    frozen: String,
}

/// The mix (futures) pending-orders response wraps the orders in `entrustedList`.
#[derive(Deserialize)]
struct RawMixPending {
    #[serde(rename = "entrustedList", default)]
    entrusted_list: Vec<RawOrder>,
}

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

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

impl Bitget {
    /// Open positions on the USDⓈ-M futures account (`/api/v2/mix/position/all-position`).
    ///
    /// # 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/v2/mix/position/all-position",
            "productType=USDT-FUTURES&marginCoin=USDT",
            "",
        )?;
        let list: Vec<RawBitgetPosition> = parse_json(data)?;
        list.iter()
            .filter(|p| symbol.is_none_or(|s| p.symbol == Self::wire_symbol(s)))
            .filter_map(parse_bitget_position)
            .collect()
    }

    /// Set the leverage for `symbol` (`/api/v2/mix/account/set-leverage`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the leverage is rejected or the request fails.
    pub fn set_leverage(&self, symbol: &Symbol, leverage: u32) -> Result<()> {
        let body = serde_json::json!({
            "symbol": Self::wire_symbol(symbol),
            "productType": "USDT-FUTURES",
            "marginCoin": "USDT",
            "leverage": leverage.to_string(),
        });
        self.signed_request(
            HttpMethod::Post,
            "/api/v2/mix/account/set-leverage",
            "",
            &body.to_string(),
        )?;
        Ok(())
    }

    /// Set the margin mode for `symbol` (`/api/v2/mix/account/set-margin-mode`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the change is rejected or the request fails.
    pub fn set_margin_mode(&mut self, symbol: &Symbol, mode: MarginMode) -> Result<()> {
        let margin = margin_wire(mode);
        let body = serde_json::json!({
            "symbol": Self::wire_symbol(symbol),
            "productType": "USDT-FUTURES",
            "marginCoin": "USDT",
            "marginMode": margin,
        });
        self.signed_request(
            HttpMethod::Post,
            "/api/v2/mix/account/set-margin-mode",
            "",
            &body.to_string(),
        )?;
        // Every futures order carries `marginMode`, so the account-level change
        // above is only half of it: without this the next order still says
        // `crossed` and overrides what was just set.
        self.margin_mode = mode;
        Ok(())
    }

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

    /// Place several orders on one symbol in one request
    /// (`/api/v2/spot/trade/batch-orders`). Bitget returns separate success and
    /// failure lists, so a synthetic `clientOid` per index re-aligns them to each
    /// request's own [`Result`]. Batch requires a single symbol (that of the
    /// first request).
    ///
    /// # Errors
    /// Returns an [`Error`] if the batch request itself fails, or if called on a
    /// futures client (mix batch is a documented follow-up).
    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("Bitget"));
        }
        for request in requests {
            self.ensure_reduce_only_is_reducible(request)?;
        }
        if requests.is_empty() {
            return Ok(Vec::new());
        }
        let wire = Self::wire_symbol(&requests[0].symbol);
        // Resolved before the batch is built, so a request the venue cannot
        // express refuses the whole call rather than being sent weakened.
        let forces = requests.iter().map(force_for).collect::<Result<Vec<_>>>()?;
        let order_list: Vec<serde_json::Value> = requests
            .iter()
            .zip(&forces)
            .map(|(r, force)| {
                let coid = self.batch_client_oid(r);
                let mut o = serde_json::json!({
                    "side": side_str(r.side),
                    "orderType": order_type_str(r.order_type),
                    "force": force,
                    "size": format_decimal(r.quantity),
                    "clientOid": coid,
                });
                if let Some(mode) = stp_mode_str(r.stp) {
                    o["stpMode"] = serde_json::json!(mode);
                }
                if let Some(price) = r.price {
                    o["price"] = serde_json::json!(format_decimal(price));
                }
                if self.is_futures() {
                    // The position half of the order, which this builder was
                    // dropping entirely: a batched reduce-only order lost the
                    // flag, and a hedged one never said which side it acts on.
                    if self.position_mode == PositionMode::Hedge {
                        o["tradeSide"] = serde_json::json!(trade_side(r));
                    } else if r.reduce_only {
                        o["reduceOnly"] = serde_json::json!("YES");
                    }
                }
                o
            })
            .collect();
        let (path, body) = if self.is_futures() {
            (
                "/api/v2/mix/order/batch-place-order",
                serde_json::json!({
                    "symbol": wire,
                    "productType": "USDT-FUTURES",
                    "marginMode": margin_wire(self.margin_mode),
                    "marginCoin": "USDT",
                    "orderList": order_list,
                }),
            )
        } else {
            (
                "/api/v2/spot/trade/batch-orders",
                serde_json::json!({ "symbol": wire, "orderList": order_list }),
            )
        };
        let data = self.signed_request(HttpMethod::Post, path, "", &body.to_string())?;
        let result: BatchOrdersResult = parse_json(data)?;
        let ok_map: HashMap<String, String> = result
            .success_list
            .into_iter()
            .map(|s| (s.client_oid, s.order_id))
            .collect();
        let fail_map: HashMap<String, String> = result
            .failure_list
            .into_iter()
            .map(|f| (f.client_oid, f.error_msg))
            .collect();
        Ok(requests
            .iter()
            .map(|req| {
                let coid = self.batch_client_oid(req);
                if let Some(order_id) = ok_map.get(&coid) {
                    Ok(Order {
                        id: order_id.clone(),
                        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,
                    })
                } else {
                    Err(Error::OrderRejected {
                        code: "batch".to_string(),
                        message: fail_map
                            .get(&coid)
                            .cloned()
                            .unwrap_or_else(|| "order rejected in batch".to_string()),
                    })
                }
            })
            .collect())
    }

    /// Cancel several orders on one `symbol` in one request
    /// (`/api/v2/spot/trade/batch-cancel-order`).
    ///
    /// # Errors
    /// Returns an [`Error`] if the request fails, or if called on a futures client.
    pub fn cancel_batch(&self, symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        let wire = Self::wire_symbol(symbol);
        let id_list: Vec<serde_json::Value> = order_ids
            .iter()
            .map(|id| serde_json::json!({ "orderId": id }))
            .collect();
        let (path, body) = if self.is_futures() {
            (
                "/api/v2/mix/order/batch-cancel-orders",
                serde_json::json!({
                    "symbol": wire,
                    "productType": "USDT-FUTURES",
                    "marginCoin": "USDT",
                    "orderIdList": id_list,
                }),
            )
        } else {
            (
                "/api/v2/spot/trade/batch-cancel-order",
                serde_json::json!({ "symbol": wire, "orderList": id_list }),
            )
        };
        self.signed_request(HttpMethod::Post, path, "", &body.to_string())?;
        Ok(())
    }
}

impl AdvancedOrders for Bitget {
    /// Bitget spot 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: "Bitget has no in-place amend; cancel and re-place the order".to_string(),
        })
    }
    fn place_batch(&mut self, requests: &[OrderRequest]) -> Result<Vec<Result<Order>>> {
        Bitget::place_batch(self, requests)
    }
    fn cancel_batch(&mut self, symbol: &Symbol, order_ids: &[String]) -> Result<()> {
        Bitget::cancel_batch(self, symbol, order_ids)
    }
    /// Bitget has no OCO order-list (it uses standalone plan/trigger orders), so
    /// this returns an [`Error::Exchange`].
    fn place_oco(&mut self, _request: &OcoRequest) -> Result<Vec<Order>> {
        Err(Error::Exchange {
            code: "unsupported".to_string(),
            message: "Bitget has no OCO order-list; use plan/trigger orders".to_string(),
        })
    }
}

#[derive(Deserialize)]
struct BatchOrdersResult {
    #[serde(rename = "successList", default)]
    success_list: Vec<BatchSuccess>,
    #[serde(rename = "failureList", default)]
    failure_list: Vec<BatchFailure>,
}

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

#[derive(Deserialize)]
struct BatchFailure {
    #[serde(rename = "clientOid", default)]
    client_oid: String,
    #[serde(rename = "errorMsg", default)]
    error_msg: String,
}

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

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

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

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

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

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

#[derive(Deserialize)]
struct RawMixAccount {
    #[serde(rename = "marginCoin")]
    margin_coin: String,
    #[serde(default)]
    available: String,
    #[serde(default)]
    locked: String,
}

#[derive(Deserialize)]
struct RawBitgetPosition {
    symbol: String,
    #[serde(rename = "holdSide")]
    hold_side: String,
    total: String,
    #[serde(rename = "openPriceAvg", default)]
    open_price_avg: String,
    #[serde(rename = "markPrice", default)]
    mark_price: String,
    leverage: String,
    #[serde(rename = "unrealizedPL", default)]
    unrealized_pl: String,
    #[serde(rename = "marginMode", default)]
    margin_mode: String,
}

fn parse_bitget_position(raw: &RawBitgetPosition) -> Option<Result<Position>> {
    let total = match parse_decimal(&raw.total) {
        Ok(total) if !total.is_zero() => total,
        Ok(_) => return None, // flat position
        Err(e) => return Some(Err(e)),
    };
    let side = if raw.hold_side == "short" {
        PositionSide::Short
    } else {
        PositionSide::Long
    };
    let build = || -> Result<Position> {
        Ok(Position {
            symbol: split_wire_symbol(&raw.symbol),
            side,
            quantity: total,
            entry_price: parse_decimal(&raw.open_price_avg)?,
            mark_price: parse_decimal(&raw.mark_price)?,
            leverage: parse_decimal(&raw.leverage)?,
            unrealized_pnl: parse_decimal(&raw.unrealized_pl)?,
            margin_mode: if raw.margin_mode.eq_ignore_ascii_case("isolated") {
                MarginMode::Isolated
            } else {
                MarginMode::Cross
            },
        })
    };
    Some(build())
}

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

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

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

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

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

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

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

    /// A futures client with a mock socket.
    fn futures_ws_client(now_ms: i64) -> (Bitget, Arc<MockWsTransport>) {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        let bitget = Bitget::with_credentials(
            Box::new(ArcTransport(http)),
            &opts,
            Credentials::new("APIKEY", "SECRET").with_passphrase("PASS"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))))
        .with_clock(Box::new(move || now_ms));
        (bitget, ws)
    }

    /// A futures client over HTTP only, for the polled reads.
    fn futures_client() -> (Bitget, Arc<MockHttpTransport>) {
        let mock = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::UsdMFutures);
        (
            Bitget::with_http(Box::new(ArcTransport(Arc::clone(&mock))), &opts),
            mock,
        )
    }
    #[test]
    fn a_spot_client_refuses_the_derivatives_surface() {
        let (mut bitget, _) = client();
        for channel in [
            DerivativesChannel::Funding,
            DerivativesChannel::MarkIndex,
            DerivativesChannel::Liquidations,
        ] {
            let err = Bitget::subscribe_derivatives(&mut bitget, &symbol(), channel)
                .expect_err("spot has no derivatives channels");
            assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        }
        let (bitget, _) = client();
        assert!(Bitget::open_interest(&bitget, &symbol()).is_err());
        assert!(Bitget::long_short_ratio(&bitget, &symbol()).is_err());
    }

    /// Bitget publishes no public stream of forced orders, so the subscription
    /// is refused rather than accepted and left to deliver nothing.
    #[test]
    fn liquidations_are_refused_because_bitget_publishes_none() {
        let (mut bitget, ws) = futures_ws_client(1_700_000_000_000);
        let err =
            Bitget::subscribe_derivatives(&mut bitget, &symbol(), DerivativesChannel::Liquidations)
                .expect_err("Bitget has no public liquidation stream");
        assert!(matches!(err, Error::Exchange { ref code, .. } if code == "unsupported"));
        assert!(ws.sent().is_empty(), "refused, yet subscribed to something");
    }

    /// One frame answers both subscriptions, because the venue reads all three
    /// prices at one moment and sends them together.
    #[test]
    fn the_ticker_frame_answers_funding_and_mark_index_together() {
        let (mut bitget, ws) = futures_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"action":"snapshot","arg":{"instType":"USDT-FUTURES","channel":"ticker",
            "instId":"BTCUSDT"},"data":[{"instId":"BTCUSDT","lastPr":"81115.8","bidPr":"81115.7",
            "askPr":"81115.9","baseVolume":"12345","markPrice":"81117","indexPrice":"81155.1985",
            "fundingRate":"0.000039","ts":"1788459657009"}]}"#
                .to_string(),
        ))]);
        Bitget::subscribe_derivatives(&mut bitget, &symbol(), DerivativesChannel::Funding).unwrap();
        Bitget::subscribe_derivatives(&mut bitget, &symbol(), DerivativesChannel::MarkIndex)
            .unwrap();

        let events = Bitget::poll_events(&mut bitget);
        assert_eq!(
            events.len(),
            3,
            "ticker, funding and mark/index: {events:?}"
        );

        let Event::Derivatives(DerivativesFeed::Funding(funding)) = &events[1] else {
            panic!("expected a funding print, got {:?}", events[1]);
        };
        assert_eq!(funding.rate, dec!(0.000039));
        assert_eq!(funding.mark_price, dec!(81117));
        assert_eq!(funding.timestamp, 1_788_459_657_009);

        let Event::Derivatives(DerivativesFeed::MarkIndex(mark)) = &events[2] else {
            panic!("expected a mark/index print, got {:?}", events[2]);
        };
        assert_eq!(mark.mark_price, dec!(81117));
        assert_eq!(mark.index_price, dec!(81155.1985));
    }

    /// A caller watching prices does not start receiving funding prints.
    #[test]
    fn an_unsubscribed_derivatives_print_is_not_emitted() {
        let (mut bitget, ws) = futures_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"action":"snapshot","arg":{"instType":"USDT-FUTURES","channel":"ticker",
            "instId":"BTCUSDT"},"data":[{"instId":"BTCUSDT","lastPr":"81115.8","bidPr":"81115.7",
            "askPr":"81115.9","baseVolume":"12345","markPrice":"81117","indexPrice":"81155.1985",
            "fundingRate":"0.000039","ts":"1788459657009"}]}"#
                .to_string(),
        ))]);
        Bitget::subscribe_ticker(&mut bitget, &symbol()).unwrap();

        let events = Bitget::poll_events(&mut bitget);
        assert_eq!(events.len(), 1, "only the ticker was asked for: {events:?}");
        assert!(matches!(events[0], Event::Ticker(_)));
    }

    #[test]
    fn open_interest_reads_the_base_currency_figure() {
        let (bitget, mock) = futures_client();
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","requestTime":1788459644402,
            "data":{"openInterestList":[{"symbol":"BTCUSDT","size":"34083.9067999999127"}],
            "ts":"1788459644403"}}"#,
        );
        let open_interest = Bitget::open_interest(&bitget, &symbol()).unwrap();
        assert_eq!(open_interest.open_interest, dec!(34083.9067999999127));
        assert_eq!(open_interest.timestamp, 1_788_459_644_403);
        assert!(mock.recorded_requests()[0]
            .url
            .contains("productType=USDT-FUTURES"));
    }

    /// Bitget publishes both proportions directly, so nothing is derived.
    #[test]
    fn the_long_short_ratio_is_read_not_derived() {
        let (bitget, mock) = futures_client();
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","requestTime":1788459645484,
            "data":[{"longAccountRatio":"0.5236","shortAccountRatio":"0.4764",
                     "longShortAccountRatio":"1.099","ts":"1788450600000"},
                    {"longAccountRatio":"0.5255","shortAccountRatio":"0.4745",
                     "longShortAccountRatio":"1.107","ts":"1788450000000"}]}"#,
        );
        let ratio = Bitget::long_short_ratio(&bitget, &symbol()).unwrap();
        assert_eq!(ratio.long_size, dec!(0.5236));
        assert_eq!(ratio.short_size, dec!(0.4764));
        // Newest first: the older row must not be the one read.
        assert_eq!(ratio.timestamp, 1_788_450_600_000);
    }

    /// Every WebSocket subscription names the market the client is on.
    ///
    /// Bitget v2 serves one socket URL for every product and tells them apart by
    /// `instType`, the way the REST paths tell them apart by `productType`. Both
    /// subscribe paths hardcoded `SPOT`, so a futures client streamed the spot
    /// book and the spot trades while its REST calls read futures -- and its
    /// private stream watched the spot account, where a futures order never
    /// appears. Nothing failed: the venue answers a spot subscription perfectly
    /// well, with the wrong market's data.
    #[test]
    fn a_futures_client_subscribes_to_futures_streams_not_spot() {
        let (mut bitget, ws) = futures_ws_client(1_700_000_000_000);
        Bitget::subscribe_trades(&mut bitget, &symbol()).unwrap();
        Bitget::subscribe_book(&mut bitget, &symbol()).unwrap();

        for frame in ws.sent() {
            assert!(
                frame.contains(r#""instType":"USDT-FUTURES""#),
                "a futures client subscribed to something else: {frame}"
            );
            assert!(!frame.contains(r#""instType":"SPOT""#));
        }

        // And the private stream, which is where the wrong market is quietest:
        // a spot account simply never reports a futures order.
        let (mut bitget, ws) = futures_ws_client(1_700_000_000_000);
        ws.push_connection(vec![Ok(Some(
            r#"{"event":"login","code":"0"}"#.to_string(),
        ))]);
        Bitget::subscribe_user_data(&mut bitget).unwrap();
        let subscribe = ws
            .sent()
            .into_iter()
            .find(|frame| frame.contains("\"channel\":\"orders\""))
            .expect("the user-data stream subscribes to orders");
        assert!(
            subscribe.contains(r#""instType":"USDT-FUTURES""#),
            "{subscribe}"
        );
    }

    /// The spot client is unchanged, which is the other half of the contract.
    #[test]
    fn a_spot_client_still_subscribes_to_spot_streams() {
        let (mut bitget, ws) = signed_ws_client(1_700_000_000_000);
        Bitget::subscribe_trades(&mut bitget, &symbol()).unwrap();
        assert!(ws.sent()[0].contains(r#""instType":"SPOT""#));
    }

    #[test]
    fn subscribe_user_data_logs_in_and_streams_orders_and_account() {
        let (mut bitget, ws) = signed_ws_client(1_700_000_000_000);
        ws.push_connection(vec![
            Ok(Some(r#"{"event":"login","code":"0"}"#.to_string())),
            Ok(Some(r#"{"event":"subscribe"}"#.to_string())),
            Ok(Some(
                r#"{"arg":{"instType":"SPOT","channel":"orders","instId":"default"},"data":[
                {"instId":"BTCUSDT","orderId":"55","clientOid":"my","side":"buy","orderType":"limit",
                "status":"filled","size":"1","baseVolume":"1","price":"100","priceAvg":"100"}]}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"arg":{"instType":"SPOT","channel":"account","coin":"default"},"data":[
                {"coin":"USDT","available":"900","frozen":"50"}]}"#
                    .to_string(),
            )),
        ]);
        bitget.subscribe_user_data().unwrap();
        assert_eq!(ws.connected_urls()[0], "wss://ws.bitget.com/v2/ws/private");
        assert!(ws.sent()[0].contains(r#""op":"login""#));
        assert!(ws.sent()[0].contains(r#""apiKey":"APIKEY""#));
        assert!(ws.sent()[1].contains(r#""channel":"orders""#));
        assert!(ws.sent()[1].contains(r#""channel":"account""#));

        let events = bitget.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.symbol, symbol());
        assert_eq!(order.side, OrderSide::Buy);
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.average_price, Some(dec!(100)));
        let Event::BalanceUpdate(balances) = &events[1] else {
            panic!("second event must be a balance update");
        };
        assert_eq!(balances.len(), 1);
        assert_eq!(balances[0].asset, "USDT");
        assert_eq!(balances[0].free, dec!(900));
        assert_eq!(balances[0].locked, dec!(50));
    }

    #[test]
    fn subscribe_user_data_requires_a_passphrase() {
        let http = Arc::new(MockHttpTransport::new());
        let ws = Arc::new(MockWsTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut bitget = Bitget::with_credentials(
            Box::new(ArcTransport(http)),
            &opts,
            Credentials::new("APIKEY", "SECRET"),
        )
        .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        assert!(matches!(
            bitget.subscribe_user_data().unwrap_err(),
            Error::InvalidCredentials(_)
        ));
    }

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

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

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

        let events = bitget.poll_events();
        assert!(events.contains(&Event::Disconnected));
        assert!(events.contains(&Event::Reconnected));
        // Two private connections (initial + reconnect), each re-signing op:login.
        let login_frames = ws
            .sent()
            .into_iter()
            .filter(|f| f.contains(r#""op":"login""#))
            .count();
        assert_eq!(login_frames, 2);
        assert_eq!(ws.connected_urls().len(), 2);
        assert_eq!(ws.connected_urls()[1], "wss://ws.bitget.com/v2/ws/private");
    }

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

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

    /// A trigger goes to the plan-order endpoint, and names which price arms it.
    ///
    /// `triggerType` is not inferred by the venue. On futures the mark is what
    /// a position is liquidated against, so that is what a stop there watches;
    /// on spot there is no mark, so the traded price is the only honest answer.
    /// Leaving it out would take whichever default the venue happens to hold.
    #[test]
    fn a_spot_trigger_goes_to_the_plan_endpoint_and_watches_the_traded_price() {
        let (bitget, mock) = signed_client(1_700_000_000_000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"orderId":"p-1","clientOid":""}}"#,
        );
        let request = OrderRequest {
            order_type: OrderType::StopMarket,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::market_sell(symbol(), dec!(1))
        };
        Bitget::place_order(&bitget, &request).unwrap();

        let sent = &mock.recorded_requests()[0];
        assert!(
            sent.url.contains("/api/v2/spot/trade/place-plan-order"),
            "{}",
            sent.url
        );
        let body = sent.body.clone().unwrap();
        assert!(body.contains(r#""triggerPrice":"19000""#), "{body}");
        assert!(body.contains(r#""triggerType":"fill_price""#), "{body}");
    }

    #[test]
    fn a_futures_trigger_watches_the_mark_and_names_its_plan_type() {
        let (bitget, mock) = signed_futures_client(1_700_000_000_000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"orderId":"p-2","clientOid":""}}"#,
        );
        let request = OrderRequest {
            order_type: OrderType::StopLimit,
            stop_price: Some(dec!(19000)),
            ..OrderRequest::limit_sell(symbol(), dec!(1), dec!(18900))
        };
        Bitget::place_order(&bitget, &request).unwrap();

        let sent = &mock.recorded_requests()[0];
        assert!(
            sent.url.contains("/api/v2/mix/order/place-plan-order"),
            "{}",
            sent.url
        );
        let body = sent.body.clone().unwrap();
        assert!(body.contains(r#""triggerType":"mark_price""#), "{body}");
        // `normal_plan` is the ordinary trigger; the other plan types belong to
        // an already-open position rather than to this request.
        assert!(body.contains(r#""planType":"normal_plan""#), "{body}");
        assert!(body.contains(r#""executePrice":"18900""#), "{body}");
        assert!(body.contains(r#""productType":"USDT-FUTURES""#), "{body}");
    }

    #[test]
    fn a_bitget_trigger_without_a_stop_price_is_refused() {
        let (bitget, 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!(Bitget::place_order(&bitget, &request).is_err());
        assert_eq!(mock.recorded_requests(), Vec::new());
    }

    const MIX_POSITIONS: &str = r#"{"code":"00000","msg":"success","data":[
        {"symbol":"BTCUSDT","holdSide":"long","total":"0.5","openPriceAvg":"20000","markPrice":"20100","leverage":"10","unrealizedPL":"50","marginMode":"isolated"}
    ]}"#;

    #[test]
    fn stp_maps_to_stp_mode() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
        );
        bitget
            .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#""stpMode":"cancel_both""#));
    }

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

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

    #[test]
    fn post_only_maps_to_the_post_only_order_type() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
        );
        bitget
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)).post_only())
            .unwrap();
        assert!(mock.recorded_requests()[0]
            .body
            .as_deref()
            .unwrap()
            .contains("post_only"));
    }

    #[test]
    fn hedge_mode_sends_trade_side_instead_of_reduce_only() {
        let (hedged, mock) = hedged_futures_client(1000);
        for _ in 0..2 {
            mock.push_json(
                200,
                r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
            );
        }
        hedged
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        hedged
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)).reduce_only())
            .unwrap();
        let reqs = mock.recorded_requests();
        assert!(reqs[0]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""tradeSide":"open""#));
        assert!(reqs[1]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""tradeSide":"close""#));
        assert!(!reqs[1].body.as_deref().unwrap().contains("reduceOnly"));
    }

    #[test]
    fn the_batch_entry_carries_the_position_half_it_was_dropping() {
        // The batched builder sent neither `reduceOnly` nor `tradeSide`, so a
        // batched reduce-only order silently became an opening one.
        let (one_way, mock) = signed_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{
            "successList":[{"orderId":"o1","clientOid":"b-0"}],"failureList":[]}}"#,
        );
        one_way
            .place_batch(&[OrderRequest::limit_sell(symbol(), dec!(1), dec!(100))
                .reduce_only()
                .with_client_order_id("b-0")])
            .unwrap();
        assert!(mock.recorded_requests()[0]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""reduceOnly":"YES""#));

        let (hedged, hedged_mock) = hedged_futures_client(1000);
        hedged_mock.push_json(
            200,
            r#"{"code":"00000","data":{
            "successList":[{"orderId":"o1","clientOid":"b-0"}],"failureList":[]}}"#,
        );
        hedged
            .place_batch(&[OrderRequest::limit_sell(symbol(), dec!(1), dec!(100))
                .reduce_only()
                .with_client_order_id("b-0")])
            .unwrap();
        assert!(hedged_mock.recorded_requests()[0]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""tradeSide":"close""#));
    }

    #[test]
    fn configured_margin_mode_reaches_every_futures_order() {
        // Both futures order paths used to send `crossed` literally, so a client
        // configured for isolated margin traded cross -- silently, with the whole
        // account balance behind the position.
        let (isolated, mock) = isolated_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
        );
        isolated
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        mock.push_json(
            200,
            r#"{"code":"00000","data":{
            "successList":[{"orderId":"o1","clientOid":"b-0"}],"failureList":[]}}"#,
        );
        isolated
            .place_batch(&[
                OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)).with_client_order_id("b-0")
            ])
            .unwrap();

        let reqs = mock.recorded_requests();
        assert!(reqs[0]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""marginMode":"isolated""#));
        assert!(reqs[1]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""marginMode":"isolated""#));

        // The default is still cross, spelled the way Bitget spells it.
        let (cross, cross_mock) = signed_futures_client(1000);
        cross_mock.push_json(
            200,
            r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
        );
        cross
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();
        let cross_reqs = cross_mock.recorded_requests();
        assert!(cross_reqs[0]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""marginMode":"crossed""#));
    }

    #[test]
    fn set_margin_mode_changes_what_later_orders_carry() {
        // The account-level call is only half of it: the order carries the mode
        // too, and a stale one overrides what was just set.
        let (mut bitget, mock) = signed_futures_client(1000);
        mock.push_json(200, r#"{"code":"00000","data":{}}"#);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"orderId":"1","clientOid":""}}"#,
        );

        Derivatives::set_margin_mode(&mut bitget, &symbol(), MarginMode::Isolated).unwrap();
        bitget
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(100)))
            .unwrap();

        let reqs = mock.recorded_requests();
        assert!(reqs[1]
            .body
            .as_deref()
            .unwrap()
            .contains(r#""marginMode":"isolated""#));
    }

    /// Bitget stamps both public reads. Payloads captured from the live
    /// endpoints.
    #[test]
    fn the_public_reads_carry_the_venue_stamp() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":[{"symbol":"BTCUSDT","lastPr":"77100.52","bidPr":"77100.51","askPr":"77100.52","baseVolume":"2886.70","ts":"1788396650211"}]}"#,
        );
        let ticker = bitget.ticker(&symbol()).unwrap();
        assert_eq!(ticker.timestamp, 1_788_396_650_211);

        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"asks":[["77100.52","0.56"]],"bids":[["77100.51","0.37"]],"ts":"1788396651599"}}"#,
        );
        let book = bitget.order_book(&symbol(), 5).unwrap();
        assert_eq!(book.timestamp, 1_788_396_651_599);
    }

    #[test]
    fn place_batch_aligns_success_and_failure() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{
            "successList":[{"orderId":"o1","clientOid":"wbatch-0"}],
            "failureList":[{"clientOid":"wbatch-1","errorMsg":"insufficient"}]}}"#,
        );
        let results = bitget
            .place_batch(&[
                // Named explicitly: this test is about aligning the venue's
                // success/failure lists back onto the requests, not about which
                // id the generator picks.
                OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
                    .with_client_order_id("wbatch-0"),
                OrderRequest::limit_buy(symbol(), dec!(1), dec!(101))
                    .with_client_order_id("wbatch-1"),
            ])
            .unwrap();
        assert_eq!(results[0].as_ref().unwrap().id, "o1");
        assert!(matches!(
            results[1].as_ref().unwrap_err(),
            Error::OrderRejected { .. }
        ));
        let reqs = mock.recorded_requests();
        assert!(reqs[0].url.contains("/api/v2/spot/trade/batch-orders"));
        assert!(reqs[0]
            .body
            .as_ref()
            .unwrap()
            .contains(r#""clientOid":"wbatch-0""#));
    }

    #[test]
    fn cancel_batch_is_one_call() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(200, r#"{"code":"00000","data":{"successList":[]}}"#);
        bitget
            .cancel_batch(&symbol(), &["1".to_string(), "2".to_string()])
            .unwrap();
        let reqs = mock.recorded_requests();
        assert_eq!(reqs.len(), 1);
        assert!(reqs[0]
            .url
            .contains("/api/v2/spot/trade/batch-cancel-order"));
    }

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

    #[test]
    fn futures_query_order_uses_mix_detail_and_state() {
        let (bitget, mock) = signed_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"symbol":"BTCUSDT","orderId":"88","clientOid":"",
            "side":"buy","orderType":"limit","state":"filled","size":"2","baseVolume":"2",
            "price":"100","priceAvg":"100"}}"#,
        );
        let order = bitget.query_order(&symbol(), "88").unwrap();
        assert_eq!(order.id, "88");
        assert_eq!(order.status, OrderStatus::Filled);
        let url = &mock.recorded_requests()[0].url;
        assert!(url.contains("/api/v2/mix/order/detail"));
        assert!(url.contains("productType=USDT-FUTURES"));
    }

    #[test]
    fn futures_cancel_and_open_orders_use_mix_endpoints() {
        let (bitget, mock) = signed_futures_client(1000);
        mock.push_json(200, r#"{"code":"00000","data":{"orderId":"88"}}"#);
        bitget.cancel_order(&symbol(), "88").unwrap();
        mock.push_json(
            200,
            r#"{"code":"00000","data":{"entrustedList":[{"symbol":"BTCUSDT","orderId":"90",
            "clientOid":"","side":"sell","orderType":"limit","state":"live","size":"3",
            "baseVolume":"0","price":"21000","priceAvg":"0"}]}}"#,
        );
        let orders = bitget.open_orders(Some(&symbol())).unwrap();
        assert_eq!(orders.len(), 1);
        assert_eq!(orders[0].side, OrderSide::Sell);
        let reqs = mock.recorded_requests();
        assert!(reqs[0].url.contains("/api/v2/mix/order/cancel-order"));
        assert!(reqs[1].url.contains("/api/v2/mix/order/orders-pending"));
    }

    #[test]
    fn futures_place_batch_uses_mix_endpoint() {
        let (bitget, mock) = signed_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":{
            "successList":[{"orderId":"o1","clientOid":"wbatch-0"}],
            "failureList":[]}}"#,
        );
        let results = bitget
            .place_batch(&[OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))
                .with_client_order_id("wbatch-0")])
            .unwrap();
        assert_eq!(results[0].as_ref().unwrap().id, "o1");
        let req = &mock.recorded_requests()[0];
        assert!(req.url.contains("/api/v2/mix/order/batch-place-order"));
        assert!(req
            .body
            .as_ref()
            .unwrap()
            .contains(r#""productType":"USDT-FUTURES""#));
    }

    #[test]
    fn futures_ticker_uses_mix_path() {
        let (bitget, mock) = signed_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":[{"symbol":"BTCUSDT","lastPr":"20000","bidPr":"19999","askPr":"20001","baseVolume":"1000"}]}"#,
        );
        let ticker = bitget.ticker(&symbol()).unwrap();
        assert_eq!(ticker.last, dec!(20000));
        let url = &mock.recorded_requests()[0].url;
        assert!(url.contains("/api/v2/mix/market/ticker"));
        assert!(url.contains("productType=USDT-FUTURES"));
    }

    #[test]
    fn futures_place_order_uses_mix_path() {
        let (bitget, mock) = signed_futures_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"orderId":"9","clientOid":""}}"#,
        );
        bitget
            .place_order(&OrderRequest::limit_buy(symbol(), dec!(1), dec!(20000)))
            .unwrap();
        let req = &mock.recorded_requests()[0];
        assert!(req.url.contains("/api/v2/mix/order/place-order"));
        assert!(req.body.as_deref().unwrap().contains("USDT-FUTURES"));
    }

    #[test]
    fn derivatives_positions_parse() {
        let (mut bitget, mock) = signed_futures_client(1000);
        mock.push_json(200, MIX_POSITIONS);
        let positions = Derivatives::positions(&mut bitget, 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!(0.5));
        assert_eq!(positions[0].leverage, dec!(10));
        assert_eq!(positions[0].margin_mode, MarginMode::Isolated);
        assert!(mock.recorded_requests()[0]
            .url
            .contains("/api/v2/mix/position/all-position"));
    }

    #[test]
    fn derivatives_set_leverage_hits_endpoint() {
        let (mut bitget, mock) = signed_futures_client(1000);
        mock.push_json(200, r#"{"code":"00000","msg":"success","data":{}}"#);
        Derivatives::set_leverage(&mut bitget, &symbol(), 20).unwrap();
        let req = &mock.recorded_requests()[0];
        assert!(req.url.contains("/api/v2/mix/account/set-leverage"));
        assert!(req.body.as_deref().unwrap().contains(r#""leverage":"20""#));
    }

    #[test]
    fn derivatives_close_position_reduce_only() {
        let (mut bitget, mock) = signed_futures_client(1000);
        mock.push_json(200, MIX_POSITIONS);
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"orderId":"9","clientOid":""}}"#,
        );
        Derivatives::close_position(&mut bitget, &symbol()).unwrap();
        let req = &mock.recorded_requests()[1];
        assert!(req.url.contains("/api/v2/mix/order/place-order"));
        let body = req.body.as_deref().unwrap();
        assert!(body.contains(r#""side":"sell""#));
        assert!(body.contains(r#""reduceOnly":"YES""#));
    }

    #[test]
    fn ticker_and_error_mapping() {
        let (bitget, mock) = client();
        mock.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":[{"symbol":"BTCUSDT","lastPr":"20000",
            "bidPr":"19999","askPr":"20001","baseVolume":"1234"}]}"#,
        );
        let ticker = bitget.ticker(&symbol()).unwrap();
        assert_eq!(ticker.last, dec!(20000));
        assert_eq!(ticker.bid, dec!(19999));

        let (bitget, mock) = client();
        mock.push_json(200, r#"{"code":"43012","msg":"insufficient","data":null}"#);
        assert!(matches!(
            bitget.ticker(&symbol()).unwrap_err(),
            Error::InsufficientBalance
        ));
    }

    #[test]
    fn klines_chronological() {
        let (bitget, mock) = client();
        mock.push_json(
            200,
            r#"{"code":"00000","data":[["1700000000000","100","110","95","105","1","0"],
            ["1700000060000","105","106","104","105.5","2","0"]]}"#,
        );
        let candles = bitget.klines(&symbol(), "1m", 2).unwrap();
        assert_eq!(candles[0].timestamp, 1_700_000_000_000);
        assert_eq!(candles[1].timestamp, 1_700_000_060_000);
    }

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

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

        let req = &mock.recorded_requests()[0];
        assert_eq!(req.method, HttpMethod::Post);
        let header = |name: &str| {
            req.headers
                .iter()
                .find(|(k, _)| k == name)
                .map(|(_, v)| v.as_str())
                .unwrap()
        };
        let ts = header("ACCESS-TIMESTAMP");
        assert_eq!(ts, "1000");
        let body = req.body.as_ref().unwrap();
        let prehash = format!("{ts}POST/api/v2/spot/trade/place-order{body}");
        let expected = hmac_sha256_base64(b"SECRET", prehash.as_bytes());
        assert_eq!(header("ACCESS-SIGN"), expected);
        assert_eq!(header("ACCESS-KEY"), "APIKEY");
        assert_eq!(header("ACCESS-PASSPHRASE"), "PASS");
    }

    #[test]
    fn query_and_balances_and_open_orders() {
        let (bitget, mock) = signed_client(1000);
        mock.push_json(
            200,
            r#"{"code":"00000","data":[{"symbol":"BTCUSDT","orderId":"42","clientOid":"",
            "side":"sell","orderType":"market","status":"filled","size":"2","baseVolume":"2",
            "price":"0","priceAvg":"100"}]}"#,
        );
        let order = bitget.query_order(&symbol(), "42").unwrap();
        assert_eq!(order.status, OrderStatus::Filled);
        assert_eq!(order.average_price, Some(dec!(100)));
        assert_eq!(order.price, None);

        mock.push_json(
            200,
            r#"{"code":"00000","data":[{"coin":"USDT","available":"100.5","frozen":"25.5"}]}"#,
        );
        let bals = bitget.balances().unwrap();
        assert_eq!(bals[0].total(), dec!(126));

        mock.push_json(
            200,
            r#"{"code":"00000","data":[{"symbol":"ETHUSDT","orderId":"7","clientOid":"",
            "side":"buy","orderType":"limit","status":"live","size":"1","baseVolume":"0",
            "price":"50","priceAvg":"0"}]}"#,
        );
        let open = bitget.open_orders(None).unwrap();
        assert_eq!(open[0].symbol, Symbol::new("ETH", "USDT"));
    }

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

    #[test]
    fn ws_parses_trade_and_book() {
        let ws = Arc::new(MockWsTransport::new());
        ws.push_connection(vec![
            Ok(Some(
                r#"{"action":"snapshot","arg":{"instType":"SPOT","channel":"trade","instId":"BTCUSDT"},
                "data":[{"ts":"1","price":"100","size":"0.5","side":"buy"}]}"#
                    .to_string(),
            )),
            Ok(Some(
                r#"{"action":"snapshot","arg":{"instType":"SPOT","channel":"books","instId":"BTCUSDT"},
                "data":[{"ts":"5","bids":[["100","1"]],"asks":[["101","2"]]}]}"#
                    .to_string(),
            )),
        ]);
        let http = Arc::new(MockHttpTransport::new());
        let opts = ExchangeOptions::mainnet(crate::MarketType::Spot);
        let mut bitget = Bitget::with_http(Box::new(ArcTransport(http)), &opts)
            .with_ws(Box::new(ArcWs(Arc::clone(&ws))));
        bitget.subscribe_trades(&symbol()).unwrap();
        assert!(ws.sent()[0].contains(r#""channel":"trade""#));

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

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

        // Shape verified against the live public endpoint on 2026-09-01.
        http.push_json(
            200,
            r#"{"code":"00000","msg":"success","data":{"serverTime":"1004500"}}"#,
        );
        let offset = bitget.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":"00000","data":[]}"#);
        bitget.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 == "ACCESS-TIMESTAMP")
            .map(|(_, value)| value.as_str())
            .expect("signed request carries ACCESS-TIMESTAMP");
        assert_eq!(header, "1004500", "the venue's time, not ours");
    }

    /// Unnamed batch orders get ids that are unique *across* batches, not only
    /// within one.
    ///
    /// The generated id used to be the order's index in its batch, so the first
    /// order of every batch was `wbatch-0`. Bitget deduplicates on the client
    /// id, so the second batch's first order looked like a repeat of the first
    /// batch's -- and the venue answered accordingly.
    #[test]
    fn batch_client_ids_do_not_repeat_across_batches() {
        let (bitget, mock) = signed_client(1000);
        let empty = r#"{"code":"00000","data":{"successList":[],"failureList":[]}}"#;
        mock.push_json(200, empty);
        mock.push_json(200, empty);

        for _ in 0..2 {
            let _ = bitget.place_batch(&[OrderRequest::limit_buy(symbol(), dec!(1), dec!(100))]);
        }

        let bodies: Vec<String> = mock
            .recorded_requests()
            .iter()
            .filter_map(|request| request.body.clone())
            .collect();
        assert_eq!(bodies.len(), 2, "two batches were sent");
        assert_ne!(
            bodies[0], bodies[1],
            "the second batch must not reuse the first batch's client id"
        );
    }

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