rustrade-execution 0.16.0

//! Shared Binance execution infrastructure.
//!
//! Exchange-agnostic building blocks used by both the spot and margin clients:
//! reconnect/backoff, rate-limit tracking, the connection-manager stream guard, Binance error
//! parsing/classification, and the Binance-string → rustrade-enum parsers (side/order-kind/TIF).
//!
//! Nothing here is spot- or margin-specific: the parsers operate on rustrade's own types
//! (`OrderKind`, `TimeInForce`, `Side`) or on Binance's stable wire strings, so both clients
//! reuse them unchanged. The REST order-response converter names the field subset it reads as a
//! trait, [`BinanceOrderFields`], which is identical across every SDK type implementing it even
//! though the structs around it are not. The WebSocket user-data `executionReport` is decoded into
//! this crate's own [`ExecutionReport`] instead, which reads only the keys the converter uses, so a
//! key it does not read cannot fail a report, as it can through the SDK's strictly typed models.
//! Either way one converter serves both clients and keeps a venue name out of it. The remaining
//! SDK-typed converters, which differ between spot's WS-API enums and margin's REST params,
//! deliberately stay in their respective modules.
//!
//! Event deduplication lives in [`crate::client::dedup`], shared with the other clients that
//! need it, and is re-exported here so Binance call sites keep a single import site.

use crate::{
    AccountEventKind, UnindexedAccountEvent,
    error::{
        ApiError, BorrowReject, BorrowRejectReason, ConnectivityError, OrderError, OrderField,
        PrecisionLimit, PrecisionViolation, StreamTerminationReason, UnindexedClientError,
        UnindexedOrderError,
    },
    fill_recovery::{FillRecoveryFailure, FillRecoveryGap, FillRecoveryScope},
    order::{
        Order, OrderKey, OrderKind, TimeInForce, TrailingOffsetType, UnindexedInactiveOrder,
        UnindexedOrderKey,
        id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
        request::UnindexedOrderResponseCancel,
        state::{
            Cancelled, Expired, Filled, InactiveOrderState, Open, OrderState, UnindexedOrderState,
        },
    },
    trade::{AssetFees, Trade, TradeId},
};

// Deduplication moved to `client::dedup` when Hyperliquid needed the same machinery; re-exported
// here so the spot and margin call sites keep naming one module.
pub(crate) use crate::client::dedup::{
    SharedDedupCache, dedup_key_from_event, is_duplicate, new_dedup_cache,
};
use crate::client::order_recovery::{
    GAP_RETRY_BASE_SECS, GapFailure, MAX_GAP_RETRIES, PendingFills,
};
use crate::client::reconnect_budget::ReconnectBudget;
use binance_sdk::common::{
    errors::{ConnectorError, WebsocketError},
    models::{Interval, ParamBuildError, RateLimitType, RestApiResponse, WebsocketApiRateLimit},
    websocket::Subscription,
};
use chrono::{DateTime, TimeZone, Utc};
use rust_decimal::Decimal;
use rustrade_instrument::{
    Side, asset::name::AssetNameExchange, exchange::ExchangeId,
    instrument::name::InstrumentNameExchange,
};
use serde::{
    Deserialize, Deserializer,
    de::{self, Visitor},
};
use smol_str::format_smolstr;
use std::{
    fmt,
    pin::Pin,
    str::FromStr,
    sync::{
        Arc,
        atomic::{AtomicU32, AtomicU64, Ordering},
    },
    task::{Context, Poll},
    time::Duration,
};
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace, warn};

// ---------------------------------------------------------------------------
// AbortOnDropStream — ensures connection_manager task is cleaned up
// ---------------------------------------------------------------------------

// AbortOnDropStream — new type not present upstream. Ensures connection_manager
// task is cancelled (at the next .await point) when the consumer drops the stream,
// preventing a background task leak and abandoned TCP connection.

/// Wrapper stream that aborts the connection_manager JoinHandle when dropped.
///
/// This ensures the channel and its associated task are cleaned up when the consumer drops the
/// account_stream. `abort()` cancels the task at its next `.await`: in fill recovery, a
/// reconnect, a backoff or the monitor loop. The manager therefore holds every connection that
/// has finished connecting in a [`DisconnectOnDrop`], whose drop closes the socket, and its event
/// subscription in an [`UnsubscribeOnDrop`]. Without them, binance-sdk's own drop would still
/// close the socket, but would abandon any request in flight, and the subscription's callback
/// task would outlive the stream until the SDK freed the connection. An abort during the SDK's own
/// connect, before it returns the connection, is not covered.
pub(crate) struct AbortOnDropStream<S> {
    inner: S,
    handle: tokio::task::JoinHandle<()>,
}

impl<S> AbortOnDropStream<S> {
    pub(crate) fn new(inner: S, handle: tokio::task::JoinHandle<()>) -> Self {
        Self { inner, handle }
    }
}

impl<S: futures::Stream + Unpin> futures::Stream for AbortOnDropStream<S> {
    type Item = S::Item;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        Pin::new(&mut self.inner).poll_next(cx)
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        self.inner.size_hint()
    }
}

impl<S> Drop for AbortOnDropStream<S> {
    fn drop(&mut self) {
        self.handle.abort();
    }
}

// ---------------------------------------------------------------------------
// DisconnectOnDrop — closes a binance-sdk WebSocket connection when dropped
// ---------------------------------------------------------------------------

/// A binance-sdk WebSocket API connection that can be closed.
///
/// Implemented for the spot WebSocket API and for the SDK's common one, which margin drives
/// directly. Both are cheap handles onto shared connection state.
pub(crate) trait BinanceWsConnection: Send + Sync + 'static {
    /// Close the connection: binance-sdk sends a Close frame once pending requests drain.
    fn close(&self) -> impl Future<Output = anyhow::Result<()>> + Send;
}

impl BinanceWsConnection for binance_sdk::spot::websocket_api::WebsocketApi {
    fn close(&self) -> impl Future<Output = anyhow::Result<()>> + Send {
        self.disconnect()
    }
}

impl BinanceWsConnection for Arc<binance_sdk::common::websocket::WebsocketApi> {
    async fn close(&self) -> anyhow::Result<()> {
        self.disconnect().await.map_err(anyhow::Error::msg)
    }
}

/// Owns a connected binance-sdk WebSocket connection and closes it when dropped.
///
/// binance-sdk reads and writes each socket in detached tasks, which keep it open and answer the
/// server's pings until it is closed. Dropping the SDK's last handle only sends a Close frame: it
/// abandons any request still pending, and reports nothing. `disconnect()` drains pending
/// requests first and returns the result. So a connection whose owner is dropped or aborted
/// before it disconnects, such as a connection manager aborted with its stream, still closes the
/// orderly way.
///
/// Dropping this guard spawns that disconnect on the current Tokio runtime. binance-sdk bounds
/// it: up to 30 s to drain pending requests, and 30 s overall. A failure is logged at WARN.
/// Dropped outside a runtime, it cannot spawn, and logs a WARN instead; the SDK's own drop then
/// sends the Close frame once the last handle to the connection goes. A runtime that is shutting down drops the spawned task unrun, and its
/// sockets close with it. [`disconnect`](Self::disconnect) closes the connection in place and
/// reports the result. Wrap a connection only once it has connected: disconnecting one that never
/// did makes binance-sdk log a WARN.
///
/// A disconnected connection stays closed: once `disconnect()` is called on it, binance-sdk skips
/// its own reconnect and its 23 h renewal, including a reconnect already pending at that point.
pub(crate) struct DisconnectOnDrop<W: BinanceWsConnection> {
    /// `None` once [`disconnect`](Self::disconnect) has finished.
    ws: Option<W>,
    /// Names the connection in logs, e.g. "BinanceSpot account stream".
    label: &'static str,
}

impl<W: BinanceWsConnection> DisconnectOnDrop<W> {
    pub(crate) fn new(ws: W, label: &'static str) -> Self {
        Self {
            ws: Some(ws),
            label,
        }
    }

    /// Close the connection and wait for binance-sdk to report the result.
    ///
    /// An `Err` is final: the guard does not retry it. If this future is dropped before it
    /// finishes, for example by a timeout or an abort, the guard still holds the connection, and
    /// its drop spawns the disconnect again, so binance-sdk may log the close twice.
    pub(crate) async fn disconnect(mut self) -> anyhow::Result<()> {
        let result = match &self.ws {
            Some(ws) => ws.close().await,
            None => Ok(()),
        };
        self.ws = None;
        result
    }
}

impl<W: BinanceWsConnection> std::ops::Deref for DisconnectOnDrop<W> {
    type Target = W;

    // `ws` is `None` only after `disconnect`, which consumes the guard, so no caller sees it.
    #[allow(clippy::expect_used)] // Invariant: only the consuming `disconnect` clears it
    fn deref(&self) -> &W {
        self.ws
            .as_ref()
            .expect("DisconnectOnDrop holds its connection until disconnected")
    }
}

impl<W: BinanceWsConnection> Drop for DisconnectOnDrop<W> {
    fn drop(&mut self) {
        let Some(ws) = self.ws.take() else { return };
        let label = self.label;
        match tokio::runtime::Handle::try_current() {
            Ok(runtime) => {
                debug!(
                    connection = label,
                    "closing a dropped Binance WebSocket connection"
                );
                runtime.spawn(async move {
                    if let Err(e) = ws.close().await {
                        warn!(%e, connection = label, "failed to close a dropped Binance WebSocket connection");
                    }
                });
            }
            Err(_) => warn!(
                connection = label,
                "Binance WebSocket connection dropped outside a Tokio runtime: it cannot be \
                 disconnected, so binance-sdk closes it once its last handle drops, without \
                 draining pending requests"
            ),
        }
    }
}

/// A binance-sdk WebSocket event subscription, unsubscribed when dropped.
///
/// binance-sdk runs each subscription's callback in its own task, which ends on
/// [`Subscription::unsubscribe`], or once the last handle to the connection is dropped and the SDK
/// frees its event channel. A subscription dropped without unsubscribing, as by a connection
/// manager aborted with its stream, would keep that task and everything its callback holds alive
/// until then, and for good if anything else still holds the connection.
pub(crate) struct UnsubscribeOnDrop(Option<Subscription>);

impl UnsubscribeOnDrop {
    pub(crate) fn new(subscription: Subscription) -> Self {
        Self(Some(subscription))
    }

    /// Stop the callback task now.
    pub(crate) fn unsubscribe(mut self) {
        if let Some(subscription) = self.0.take() {
            subscription.unsubscribe();
        }
    }
}

impl Drop for UnsubscribeOnDrop {
    fn drop(&mut self) {
        if let Some(subscription) = self.0.take() {
            subscription.unsubscribe();
        }
    }
}

// ---------------------------------------------------------------------------
// Constants
// ---------------------------------------------------------------------------

/// If no WS activity (messages, ping, pong) for this duration, force reconnect.
pub(crate) const HEARTBEAT_TIMEOUT_SECS: u64 = 30;
// Compile-time guard: reconnect sites cast this to i64 (`chrono::Duration::seconds`); ensure it
// never overflows.
const _: () = assert!(
    HEARTBEAT_TIMEOUT_SECS <= i64::MAX as u64,
    "HEARTBEAT_TIMEOUT_SECS overflows i64"
);
/// Timeout for fill recovery REST queries after reconnect.
pub(crate) const FILL_RECOVERY_TIMEOUT_SECS: u64 = 30;

/// How far past the start of the recovery that opens it a [`FillGap`] reaches.
///
/// Recovery starts after the live stream is subscribed, so a fill after that moment arrives live.
/// Binance stamps fills by its own clock, which may run ahead of this host's. Binance rejects a
/// signed request stamped more than `recvWindow` (5 s by default) from its clock, so a skew past
/// this margin would already fail every REST call. The overlap with live delivery it leaves is
/// seconds long, and the dedup cache absorbs it, as long as those fills have not left it by the
/// time the gap is read.
pub(crate) const GAP_END_SLACK_SECS: i64 = 10;

/// A span of one instrument's fills that a reconnect's recovery has not yet forwarded.
///
/// It runs from the disconnect to just after the recovery that opened it began
/// ([`GAP_END_SLACK_SECS`]). The live stream delivers every fill after that, so reading the gap
/// never re-reads a fill the stream already delivered, however long the gap waits.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct FillGap {
    /// The first moment of the gap, in epoch milliseconds.
    pub(crate) start_ms: i64,
    /// The last moment of the gap, in epoch milliseconds.
    pub(crate) end_ms: i64,
    /// How many reads of the gap have failed or not finished.
    failures: u32,
    /// When the gap may next be read.
    due: tokio::time::Instant,
}

impl FillGap {
    /// Whether this is the same span as `other`, whatever its retry state.
    fn same_span(&self, other: &Self) -> bool {
        self.start_ms == other.start_ms && self.end_ms == other.end_ms
    }
}

/// The fills a reconnect's recovery has not yet forwarded, as [`FillGap`]s per instrument.
///
/// A reconnect opens a gap for every instrument ([`open`](Self::open)) before reading any, so a
/// recovery that fails or times out loses nothing: each gap stays until its fills are forwarded
/// ([`recovered`](Self::recovered)), and a failed one is retried with a backoff, connected or
/// across reconnects, until it has failed [`MAX_GAP_RETRIES`] retries
/// ([`failed`](Self::failed)). A reconnect does not bring a retry forward, so a flapping
/// connection cannot use the retries up.
#[derive(Debug, Default)]
pub(crate) struct UnrecoveredFills(fnv::FnvHashMap<InstrumentNameExchange, Vec<FillGap>>);

impl PendingFills for UnrecoveredFills {
    fn pending(&self, instrument: &InstrumentNameExchange) -> bool {
        self.covers(instrument)
    }
}

impl UnrecoveredFills {
    /// Open a gap for each of `instruments`, from `disconnect_time` to just after `now`, the start
    /// of the recovery that will read it, due at once.
    ///
    /// Where the new gap overlaps one the instrument already has, it starts after it instead: the
    /// kept gap covers that part and keeps its retry schedule, so no span is read twice and a
    /// reconnect never brings a retry forward. Disconnects come in time order, so an earlier part
    /// of the new gap that no kept gap covers is from before the previous disconnect, when the
    /// stream was live.
    pub(crate) fn open(
        &mut self,
        instruments: &[InstrumentNameExchange],
        disconnect_time: DateTime<Utc>,
        now: DateTime<Utc>,
    ) {
        let start_ms = disconnect_time.timestamp_millis();
        let end_ms = (now + chrono::Duration::seconds(GAP_END_SLACK_SECS)).timestamp_millis();
        let due = tokio::time::Instant::now();
        for instrument in instruments {
            let kept = self.0.get(instrument).map_or(&[][..], Vec::as_slice);
            let start_ms = kept
                .iter()
                .filter(|kept| kept.start_ms <= end_ms && start_ms <= kept.end_ms)
                .map(|kept| kept.end_ms + 1)
                .fold(start_ms, i64::max);
            if start_ms <= end_ms {
                self.0.entry(instrument.clone()).or_default().push(FillGap {
                    start_ms,
                    end_ms,
                    failures: 0,
                    due,
                });
            }
        }
    }

    /// The gaps due by `now`, to read.
    pub(crate) fn due(&self, now: tokio::time::Instant) -> Vec<(InstrumentNameExchange, FillGap)> {
        self.0
            .iter()
            .flat_map(|(instrument, gaps)| {
                gaps.iter()
                    .filter(move |gap| gap.due <= now)
                    .map(move |gap| (instrument.clone(), *gap))
            })
            .collect()
    }

    /// Whether a gap is left on `instrument`.
    pub(crate) fn covers(&self, instrument: &InstrumentNameExchange) -> bool {
        self.0.contains_key(instrument)
    }

    /// When the next gap is due, or `None` when there is none.
    pub(crate) fn next_due(&self) -> Option<tokio::time::Instant> {
        self.0.values().flatten().map(|gap| gap.due).min()
    }

    /// Whether no gap is left.
    #[cfg(test)]
    pub(crate) fn is_empty(&self) -> bool {
        self.0.is_empty()
    }

    /// Make every gap due now, as if its retry delay had passed.
    #[cfg(test)]
    pub(crate) fn make_due(&mut self) {
        let now = tokio::time::Instant::now();
        for gap in self.0.values_mut().flatten() {
            gap.due = now;
        }
    }

    /// Record that `gap`'s fills have all been forwarded.
    pub(crate) fn recovered(&mut self, instrument: &InstrumentNameExchange, gap: &FillGap) {
        self.remove_where(instrument, |kept| kept.same_span(gap));
    }

    /// Record that a read of `gap` failed or did not finish at `now`: it is retried after a
    /// backoff, or dropped once it has failed [`MAX_GAP_RETRIES`] retries. Returns which, or
    /// `None` if the gap is no longer kept.
    pub(crate) fn failed(
        &mut self,
        instrument: &InstrumentNameExchange,
        gap: &FillGap,
        now: tokio::time::Instant,
    ) -> Option<GapFailure> {
        let gaps = self.0.get_mut(instrument)?;
        let index = gaps.iter().position(|kept| kept.same_span(gap))?;
        gaps[index].failures += 1;
        if gaps[index].failures > MAX_GAP_RETRIES {
            gaps.remove(index);
            if gaps.is_empty() {
                self.0.remove(instrument);
            }
            return Some(GapFailure::GivenUp);
        }
        let delay = Duration::from_secs(GAP_RETRY_BASE_SECS << (gaps[index].failures - 1));
        gaps[index].due = now + delay;
        Some(GapFailure::Retry(delay))
    }

    fn remove_where(
        &mut self,
        instrument: &InstrumentNameExchange,
        matches: impl Fn(&FillGap) -> bool,
    ) {
        if let Some(gaps) = self.0.get_mut(instrument) {
            gaps.retain(|gap| !matches(gap));
            if gaps.is_empty() {
                self.0.remove(instrument);
            }
        }
    }
}

/// A gap boundary in epoch milliseconds, as a time for a log line.
pub(crate) fn gap_time(ms: i64) -> DateTime<Utc> {
    Utc.timestamp_millis_opt(ms).single().unwrap_or_default()
}

/// Record that fill recovery on `venue` did not read `gap` of `instrument`, because of `reason`,
/// and report what follows: a retry, logged, or, once it has failed [`MAX_GAP_RETRIES`] retries,
/// giving the gap up, which leaves its fills undelivered. A given-up gap is logged and sent on `tx`
/// as [`AccountEventKind::FillRecoveryGaveUp`].
pub(crate) fn gap_failed(
    unrecovered: &mut UnrecoveredFills,
    tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
    venue: ExchangeId,
    instrument: &InstrumentNameExchange,
    gap: &FillGap,
    reason: FillRecoveryFailure,
) {
    let (start, end) = (gap_time(gap.start_ms), gap_time(gap.end_ms));
    match unrecovered.failed(instrument, gap, tokio::time::Instant::now()) {
        Some(GapFailure::Retry(delay)) => warn!(
            %venue,
            %instrument,
            %start,
            %end,
            retry_in_secs = delay.as_secs(),
            %reason,
            "Binance fill recovery did not read this gap, retrying it later"
        ),
        Some(GapFailure::GivenUp) => {
            error!(
                %venue,
                %instrument,
                %start,
                %end,
                retries = MAX_GAP_RETRIES,
                %reason,
                "Binance fill recovery gave up on this gap: its fills are not delivered; read them \
                 with fetch_trades"
            );
            let gave_up = FillRecoveryGap::new(
                FillRecoveryScope::Instruments(vec![instrument.clone()]),
                start,
                end,
                // The first read and every retry failed.
                MAX_GAP_RETRIES + 1,
                reason,
            );
            if tx
                .send(UnindexedAccountEvent::new(
                    venue,
                    AccountEventKind::FillRecoveryGaveUp(gave_up),
                ))
                .is_err()
            {
                debug!(%venue, "Binance fill recovery: consumer dropped before the give-up was sent");
            }
        }
        None => {}
    }
}

/// Drop the executions in `page` stamped after `end_ms`, and return whether there were any: a
/// walk over a span then has everything in it, since Binance returns executions in trade-id order,
/// which is time order. An execution without a time is kept.
pub(crate) fn drop_after<T: BinanceExecutionFields>(page: &mut Vec<T>, end_ms: i64) -> bool {
    let len = page.len();
    page.retain(|execution| execution.time().is_none_or(|time| time <= end_ms));
    page.len() < len
}
/// Timeout for the initial WebSocket API TCP+TLS handshake.
/// Without this, a network partition holds the write lock for up to 75–127 s
/// (OS TCP timeout), stalling all concurrent open_order/cancel_order callers.
pub(crate) const CONNECT_TIMEOUT_SECS: u64 = 15;
// Compile-time guard: fill-recovery sites cast this to i64 (`chrono::Duration::seconds`); ensure it
// never overflows.
const _: () = assert!(
    CONNECT_TIMEOUT_SECS <= i64::MAX as u64,
    "CONNECT_TIMEOUT_SECS overflows i64"
);
/// Extra lookback subtracted from Signal disconnect timestamps to cover Tokio scheduling
/// jitter between the actual WS close and when the monitor task records Utc::now().
/// The dedup cache absorbs any resulting duplicate fills.
pub(crate) const SIGNAL_RECOVERY_LOOKBACK_MS: i64 = 500;
/// Maximum trades per Binance REST query.
/// Stored as `usize` for direct use in `Vec::len()` comparisons; cast to `i32` at SDK call sites
/// (`MyTradesParams::limit(i32)`).
pub(crate) const BINANCE_MAX_TRADES: usize = 1000;
// Compile-time guard: SDK call sites cast this to i32; ensure it never overflows.
const _: () = assert!(
    BINANCE_MAX_TRADES <= i32::MAX as usize,
    "BINANCE_MAX_TRADES overflows i32"
);
/// Cooldown after a rate-limit refusal that carried no wait. Binance sends one with most: REST's
/// `Retry-After` on a 429 or 418, and the WS API's `retryAfter`. A WAF 403 and a refused
/// WebSocket handshake carry none the client can read.
pub(crate) const DEFAULT_RATE_LIMIT_DELAY_SECS: u64 = 10;
/// The longest rate-limit cooldown a request sleeps through before it is sent. A request facing a
/// longer one fails at once with [`CooldownActive`], reported as [`ApiError::RateLimit`] with the
/// time left, rather than hanging for a ban that can last days.
///
/// Equal to [`DEFAULT_RATE_LIMIT_DELAY_SECS`], so only a wait Binance sent can exceed it: without
/// one, every cooldown is slept through.
const MAX_COOLDOWN_SLEEP: Duration = Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS);
/// Maximum number of REST retry attempts on rate-limit errors.
pub(crate) const MAX_RATE_LIMIT_RETRIES: u32 = 3;

// ---------------------------------------------------------------------------
// WS-API user-data frames
// ---------------------------------------------------------------------------

/// A text frame from a WS-API user-data subscription (`userDataStream.subscribe*`), as
/// binance-sdk hands it to a `subscribe_on_ws_events` callback: the raw frame, unchanged.
#[derive(Debug, PartialEq)]
pub(crate) enum UserDataFrame<'a> {
    /// An RPC response, such as the subscribe acknowledgement: it carries a top-level `id`.
    Response,
    /// A pushed event, which Binance wraps as `{ "subscriptionId", "event": { "e", .. } }`.
    Event {
        /// The subscription the event belongs to; it routes isolated-margin events.
        subscription_id: Option<i64>,
        /// The inner event's `e` tag.
        event_type: &'a str,
        /// The inner event, unparsed, for the matched branch's one typed pass.
        event: &'a str,
    },
    /// A frame shape this client does not know: neither of the above, or an event whose `e` tag
    /// is missing or not a plain string. Its caller logs it through [`log_unrecognised_frame`],
    /// since a change in how Binance delivers events would otherwise drop every event silently.
    Unrecognised,
}

/// Split a WS-API user-data frame into a [`UserDataFrame`].
///
/// Reads borrowed views only, with no `serde_json::Value` DOM: this runs on every inbound frame.
/// The inner event stays an unparsed slice, and only its `e` tag is read here.
pub(crate) fn parse_user_data_frame(frame: &str) -> UserDataFrame<'_> {
    use serde_json::value::RawValue;

    #[derive(serde::Deserialize)]
    struct Envelope<'a> {
        #[serde(borrow, default)]
        id: Option<&'a RawValue>,
        #[serde(borrow, default)]
        event: Option<&'a RawValue>,
        #[serde(rename = "subscriptionId", default)]
        subscription_id: Option<i64>,
    }
    // Discriminator-only view of the inner event: reads `e` without materialising the payload.
    // Binance places `e` first, so this is cheap next to the typed pass the caller then makes on
    // the matched branch only. A manual byte scan for `"e"` would be fragile to whitespace,
    // escaping and key order, so keep the typed two-pass read.
    #[derive(serde::Deserialize)]
    struct EventTag<'a> {
        #[serde(borrow, default)]
        e: Option<&'a str>,
    }

    let Ok(envelope) = serde_json::from_str::<Envelope<'_>>(frame) else {
        return UserDataFrame::Unrecognised;
    };
    if envelope.id.is_some() {
        return UserDataFrame::Response;
    }
    let Some(event) = envelope.event else {
        return UserDataFrame::Unrecognised;
    };
    let event = event.get();
    // A missing tag, or one that cannot be borrowed (escaped, or not a string), is unrecognised
    // rather than an empty type: an unknown type is ignored quietly, and this must not be.
    let Some(event_type) = serde_json::from_str::<EventTag<'_>>(event)
        .ok()
        .and_then(|tag| tag.e)
    else {
        return UserDataFrame::Unrecognised;
    };
    UserDataFrame::Event {
        subscription_id: envelope.subscription_id,
        event_type,
        event,
    }
}

/// Log an [`UserDataFrame::Unrecognised`] frame: at `warn` for the first, and for every 1000th
/// after it with the running count, and at `trace` otherwise. So a change in delivery that makes
/// every frame unrecognised is seen at once, without a warning per frame.
///
/// `seen` is the caller's process-wide counter, one per venue: it is shared by every stream of
/// that venue and never resets, so a later stream or outage warns again only at the next
/// thousandth frame, not on its first.
pub(crate) fn log_unrecognised_frame(venue: &'static str, seen: &AtomicU64, frame: &str) {
    let count = seen.fetch_add(1, Ordering::Relaxed) + 1;
    if count == 1 || count.is_multiple_of(1000) {
        warn!(
            venue,
            count,
            frame = frame_excerpt(frame),
            "Binance WS: unrecognised user-data frame (not an RPC response, nor an event envelope \
             with a readable `e` tag), ignoring it; further ones are logged at trace, with a \
             warning every 1000th"
        );
    } else {
        trace!(
            venue,
            count,
            frame = frame_excerpt(frame),
            "Binance WS: unrecognised user-data frame, ignoring it"
        );
    }
}

/// The user-data event types one venue's handlers had no arm for, for [`log_unhandled_event`].
///
/// One process-wide instance per venue (a `static`), shared by every stream of that venue and
/// never reset, like [`log_unrecognised_frame`]'s counter.
pub(crate) struct UnhandledEvents {
    count: AtomicU64,
    /// Types already warned about. Capped at [`UnhandledEvents::MAX_TYPES`], so a stream of junk
    /// `e` tags cannot grow it without bound.
    types: parking_lot::Mutex<Vec<String>>,
}

impl UnhandledEvents {
    const MAX_TYPES: usize = 32;

    pub(crate) const fn new() -> Self {
        Self {
            count: AtomicU64::new(0),
            types: parking_lot::Mutex::new(Vec::new()),
        }
    }

    /// Count one event of `event_type`. Returns the running count, and whether to log it loudly:
    /// the first of its type (while fewer than [`Self::MAX_TYPES`] are tracked), or every 1000th
    /// event overall.
    fn record(&self, event_type: &str) -> (u64, bool) {
        let count = self.count.fetch_add(1, Ordering::Relaxed) + 1;
        let first_of_type = {
            let mut types = self.types.lock();
            if types.iter().any(|seen| seen == event_type) || types.len() >= Self::MAX_TYPES {
                false
            } else {
                types.push(event_type.to_owned());
                true
            }
        };
        (count, first_of_type || count.is_multiple_of(1000))
    }
}

/// Log a user-data event whose `e` tag the stream handler has no arm for: at `warn` for the first
/// of each type, and for every 1000th unhandled event with the running count, and at `trace`
/// otherwise.
///
/// Event types a handler knowingly ignores get their own arm; this is for the rest. Warning per
/// type means a frequent unhandled event cannot hide a rarer one: a venue that renames an event,
/// or a handler that matches the wrong name, is seen at once rather than dropped silently (while
/// fewer than [`UnhandledEvents::MAX_TYPES`] types have been seen; past that, a new type waits for
/// the next 1000th).
pub(crate) fn log_unhandled_event(
    venue: &'static str,
    seen: &UnhandledEvents,
    event_type: &str,
    event: &str,
) {
    let (count, loud) = seen.record(event_type);
    if loud {
        warn!(
            venue,
            count,
            event_type,
            event = frame_excerpt(event),
            "Binance WS: user-data event of a type this client does not handle, ignoring it; \
             further ones of this type are logged at trace, with a warning every 1000th"
        );
    } else {
        trace!(
            venue,
            count, event_type, "Binance WS: unhandled user-data event, ignoring it"
        );
    }
}

/// The first 200 characters of a frame, for a log line about it.
pub(crate) fn frame_excerpt(frame: &str) -> &str {
    frame
        .char_indices()
        .nth(200)
        .map_or(frame, |(end, _)| &frame[..end])
}

// ---------------------------------------------------------------------------
// Rate limit tracker
// ---------------------------------------------------------------------------

/// Binance's documented spot request-weight limit per minute. REST `/api` and the WS-API draw on
/// it together, per IP. A spot tracker starts from this value and replaces it with the limit each
/// WS-API response reports.
pub(crate) const SPOT_REQUEST_WEIGHT_PER_MINUTE: u32 = 6_000;
/// Binance's documented per-IP weight limit per minute for IP-weighted `/sapi` endpoints. No
/// response reports it, so a margin tracker keeps this value. The separate per-UID `/sapi` limit
/// is not tracked.
pub(crate) const SAPI_IP_WEIGHT_PER_MINUTE: u32 = 12_000;
/// Used weight, as a percentage of the per-minute limit, at which queries pause until the next
/// minute. Orders and cancels keep the remainder.
const WEIGHT_PAUSE_PERCENT: u64 = 90;
/// Added to a wait that ends at a time on Binance's clock, to cover skew between this host's clock
/// and Binance's: the next minute, when its weight counters reset, and the end of a ban the WS API
/// reports (see [`wait_until_epoch_ms`]).
pub(crate) const CLOCK_SKEW_SLACK_MS: u64 = 1_000;

/// The Binance per-minute request-weight pool a [`RateLimitTracker`] follows.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WeightPool {
    /// Spot REST `/api` and the spot WS-API, which share one per-IP pool. A REST response reports
    /// the weight used in `x-mbx-used-weight-1m`; a WS-API response reports it with the limit.
    Spot,
    /// IP-weighted margin `/sapi` endpoints, which report the weight used in
    /// `x-sapi-used-ip-weight-1m`.
    Sapi,
}

impl WeightPool {
    fn default_limit(self) -> u32 {
        match self {
            Self::Spot => SPOT_REQUEST_WEIGHT_PER_MINUTE,
            Self::Sapi => SAPI_IP_WEIGHT_PER_MINUTE,
        }
    }
}

/// What a request does, which decides the rate-limit waits it honours.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RequestKind {
    /// Places or cancels an order. Waits only for a rate-limit cooldown, so it can use the weight
    /// left once queries pause.
    Order,
    /// Reads account or order state. Also waits while the weight used is near the limit.
    Query,
    /// Reads what a reconnect's fill recovery needs, or what that recovery depends on. Like
    /// [`Order`](Self::Order), waits only for a rate-limit cooldown: the pause near the limit can
    /// outlast the recovery's time budget, which would leave its fills to a retry minutes later,
    /// or undelivered once every retry has failed.
    Essential,
}

#[derive(Default)]
struct Deadlines {
    /// Set by a rate-limit response. Every request waits for it.
    blocked_until: Option<tokio::time::Instant>,
    /// Set when the weight used nears the per-minute limit. Only queries wait for it.
    throttled_until: Option<tokio::time::Instant>,
}

/// Tracks rate-limit state across REST API calls.
///
/// Two waits, both shared by clones of the client (which share one `Arc<RateLimitTracker>`):
/// - a cooldown after Binance answered with a rate-limit error, which every request honours. It
///   lasts as long as Binance asked, uncapped: a ban lasts from minutes to days, and a request
///   sent inside one lengthens it. A request sleeps through at most [`MAX_COOLDOWN_SLEEP`] of it,
///   and fails with [`CooldownActive`] when more is left;
/// - a pause until the next minute once the weight used reaches [`WEIGHT_PAUSE_PERCENT`] of the
///   per-minute limit, which only [`RequestKind::Query`] requests honour, so the weight left
///   stays free for orders, cancels and [`RequestKind::Essential`] reads. The weight used comes
///   from successful responses ([`observe_rest`](Self::observe_rest),
///   [`observe_ws_api`](Self::observe_ws_api)); a rejected request's response is not observed,
///   so usage can only be under-counted.
pub(crate) struct RateLimitTracker {
    // parking_lot::Mutex — never poisons, consistent with SharedDedupCache
    deadlines: parking_lot::Mutex<Deadlines>,
    pool: WeightPool,
    /// The pool's per-minute weight limit.
    weight_limit: AtomicU32,
}

impl RateLimitTracker {
    pub(crate) fn new(pool: WeightPool) -> Self {
        Self {
            deadlines: parking_lot::Mutex::new(Deadlines::default()),
            pool,
            weight_limit: AtomicU32::new(pool.default_limit()),
        }
    }

    /// Sleep until the waits `kind` honours have passed. Returns immediately if none is active.
    ///
    /// # Errors
    ///
    /// [`CooldownActive`], at once or after sleeping, when the rate-limit cooldown has more than
    /// [`MAX_COOLDOWN_SLEEP`] left: the request must not be sent.
    ///
    /// Loops after waking to re-check the deadline: another task may have extended it while this
    /// task was sleeping.
    pub(crate) async fn wait_if_blocked(&self, kind: RequestKind) -> Result<(), CooldownActive> {
        loop {
            let (blocked_until, deadline) = {
                let deadlines = self.deadlines.lock();
                let deadline = match kind {
                    RequestKind::Order | RequestKind::Essential => deadlines.blocked_until,
                    // `None` orders below `Some`, so this is whichever deadline is later.
                    RequestKind::Query => deadlines.blocked_until.max(deadlines.throttled_until),
                };
                (deadlines.blocked_until, deadline)
            };
            let now = tokio::time::Instant::now();
            if let Some(remaining) = blocked_until.map(|until| until.saturating_duration_since(now))
                && too_long_to_sleep(remaining)
            {
                return Err(CooldownActive { remaining });
            }
            match deadline {
                None => return Ok(()),
                Some(until) => {
                    if until <= now {
                        return Ok(());
                    }
                    // debug! not warn! — on_rate_limited and observe_used_weight already log
                    // the event; multiple concurrent callers all waiting during recover_fills
                    // would otherwise flood the log with identical lines.
                    // as_millis() returns u128; truncation impossible (u64::MAX ms ≈ 584M years)
                    #[allow(clippy::cast_possible_truncation)]
                    let delay_ms = (until - now).as_millis() as u64;
                    debug!(delay_ms, ?kind, "Binance request held back, waiting");
                    tokio::time::sleep_until(until).await;
                }
            }
        }
    }

    /// The per-minute weight limit the pause is measured against.
    #[cfg(test)]
    pub(crate) fn weight_limit(&self) -> u32 {
        self.weight_limit.load(Ordering::Relaxed)
    }

    /// Pause queries for `pause`, as a response near the weight limit would.
    #[cfg(test)]
    pub(crate) fn throttle(&self, pause: Duration) {
        let now = tokio::time::Instant::now();
        extend_deadline(&mut self.deadlines.lock().throttled_until, now, pause);
    }

    /// How long the rate-limit cooldown has left, or `None` when none is active.
    pub(crate) fn cooldown_remaining(&self) -> Option<Duration> {
        let now = tokio::time::Instant::now();
        self.deadlines
            .lock()
            .blocked_until
            .map(|until| until.saturating_duration_since(now))
            .filter(|remaining| !remaining.is_zero())
    }

    /// Whether a rate-limit cooldown is active now.
    #[cfg(test)]
    pub(crate) fn is_blocked(&self) -> bool {
        self.cooldown_remaining().is_some()
    }

    /// Record a rate-limit refusal: hold every request back for `retry_after`, the wait Binance
    /// sent, or [`DEFAULT_RATE_LIMIT_DELAY_SECS`] when it sent none. Extends the cooldown if a
    /// longer one is already active.
    ///
    /// # Panics
    ///
    /// Panics if called outside a Tokio runtime context (`tokio::time::Instant::now()`
    /// requires an active runtime).
    pub(crate) fn on_rate_limited(&self, retry_after: Option<Duration>) {
        let delay = retry_after.unwrap_or(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS));
        let now = tokio::time::Instant::now();
        let was_blocked = extend_deadline(&mut self.deadlines.lock().blocked_until, now, delay);
        // only warn on mode entry; subsequent calls from the retry loop extend
        // the cooldown silently to avoid duplicate "entering degradation mode" lines.
        if was_blocked {
            debug!(
                delay_secs = delay.as_secs(),
                "Binance rate-limit cooldown extended"
            );
        } else {
            warn!(
                delay_secs = delay.as_secs(),
                "Binance entering rate-limit degradation mode"
            );
        }
    }

    /// Record `e`, a failed account-stream connect or subscribe, in the cooldown when it is
    /// Binance refusing it for a rate limit: a handshake refused with 429 or 418, or a WS-API
    /// rate-limit response such as `-1003`. Returns whether it was one.
    ///
    /// The SDK reports a refused handshake as text, without its `Retry-After`, so a 429 starts the
    /// default cooldown and a 418 Binance's shortest ban ([`HandshakeRefusal::retry_after`]). A
    /// WS-API response names the end of a ban, which is honoured.
    pub(crate) fn on_stream_refused(&self, e: &anyhow::Error) -> bool {
        let retry_after = if let Some(refusal) = HandshakeRefusal::of(e) {
            refusal.retry_after()
        } else if let Some(RequestFailure::RateLimited(retry_after)) = ws_response_failure(e) {
            retry_after
        } else {
            return false;
        };
        warn!(error = %format_args!("{e:#}"), "Binance refused an account-stream connect for a rate limit");
        self.on_rate_limited(retry_after);
        true
    }

    /// Wait, before an account stream connects or reconnects, for a rate-limit cooldown short
    /// enough to sleep through ([`MAX_COOLDOWN_SLEEP`]), as a request would.
    ///
    /// # Errors
    /// [`StreamTerminationReason::RateLimited`] with the time left, when the cooldown is longer:
    /// connecting during it would be refused, and could lengthen a ban.
    pub(crate) async fn before_stream_connect(&self) -> Result<(), StreamTerminationReason> {
        self.wait_if_blocked(RequestKind::Essential)
            .await
            .map_err(|cooldown| end_for_cooldown(cooldown.remaining))
    }

    /// Wait, before an account stream is opened, for a rate-limit cooldown short enough to sleep
    /// through, as [`before_stream_connect`](Self::before_stream_connect) does for a reconnect.
    ///
    /// # Errors
    /// [`ApiError::RateLimit`] with the time left, when the cooldown is longer. Nothing was sent.
    pub(crate) async fn before_stream_open(&self) -> Result<(), UnindexedClientError> {
        self.wait_if_blocked(RequestKind::Essential)
            .await
            .map_err(|cooldown| {
                UnindexedClientError::Api(ApiError::rate_limited(Some(cooldown.remaining)))
            })
    }

    /// The error opening an account stream fails with, when its connect or subscribe failed with
    /// `e`. A rate-limit refusal is recorded ([`on_stream_refused`](Self::on_stream_refused)) and
    /// reported as [`ApiError::RateLimit`] with the cooldown's time left; any other failure as
    /// `otherwise` makes it.
    pub(crate) fn stream_open_error(
        &self,
        e: &anyhow::Error,
        otherwise: impl FnOnce(&anyhow::Error) -> UnindexedClientError,
    ) -> UnindexedClientError {
        if self.on_stream_refused(e) {
            UnindexedClientError::Api(ApiError::rate_limited(self.cooldown_remaining()))
        } else {
            otherwise(e)
        }
    }

    /// How long the cooldown has left, when that is longer than a request sleeps through.
    fn too_long_to_wait(&self) -> Option<Duration> {
        self.cooldown_remaining()
            .filter(|remaining| too_long_to_sleep(*remaining))
    }

    /// Record the weight a successful REST response reports as used, for a request sent at
    /// `sent_ms` (from [`unix_ms`]).
    ///
    /// Reads the header of this tracker's [`WeightPool`]; a response without it changes nothing.
    pub(crate) fn observe_rest<D>(&self, response: &RestApiResponse<D>, sent_ms: u128) {
        let used = match self.pool {
            // The SDK has already parsed `x-mbx-used-weight-1m` into `rate_limits`.
            WeightPool::Spot => response
                .rate_limits
                .iter()
                .flatten()
                .find(|limit| {
                    is_weight_per_minute(
                        &limit.rate_limit_type,
                        &limit.interval,
                        limit.interval_num,
                    )
                })
                .map(|limit| limit.count),
            // The SDK parses only `x-mbx-*` headers. reqwest names headers in lowercase.
            WeightPool::Sapi => response
                .headers
                .get("x-sapi-used-ip-weight-1m")
                .and_then(|used| used.parse().ok()),
        };
        if let Some(used) = used {
            self.observe_used_weight(used, sent_ms);
        }
    }

    /// Record the weight limit and the weight used that a WS-API response reports, for a request
    /// sent at `sent_ms` (from [`unix_ms`]).
    ///
    /// The WS-API draws on the spot pool, so a margin tracker ignores these.
    pub(crate) fn observe_ws_api(
        &self,
        rate_limits: Option<&[WebsocketApiRateLimit]>,
        sent_ms: u128,
    ) {
        debug_assert_eq!(self.pool, WeightPool::Spot, "WS-API usage is spot weight");
        if self.pool != WeightPool::Spot {
            return;
        }
        let Some(weight) = rate_limits.into_iter().flatten().find(|limit| {
            is_weight_per_minute(&limit.rate_limit_type, &limit.interval, limit.interval_num)
        }) else {
            return;
        };
        // A limit of 0 would pause every query; keep the last good value instead. It rarely
        // changes, so it is only written when it does.
        let previous = self.weight_limit.load(Ordering::Relaxed);
        if weight.limit > 0 && weight.limit != previous {
            self.weight_limit.store(weight.limit, Ordering::Relaxed);
            debug!(
                previous,
                limit = weight.limit,
                "Binance request-weight limit per minute updated"
            );
        }
        self.observe_used_weight(weight.count, sent_ms);
    }

    /// Pause queries until the end of the minute the request was sent in, `sent_ms`, if `used`
    /// has reached [`WEIGHT_PAUSE_PERCENT`] of the limit.
    ///
    /// Binance counts a request in the minute it receives it, which is the minute it was sent in
    /// barring clock skew and a request in flight across the boundary. A response observed after
    /// that minute has ended describes a counter that has since reset, so it pauses nothing,
    /// rather than holding queries back for the whole of the next minute.
    fn observe_used_weight(&self, used: u32, sent_ms: u128) {
        let limit = self.weight_limit.load(Ordering::Relaxed);
        if u64::from(used) * 100 < u64::from(limit) * WEIGHT_PAUSE_PERCENT {
            return;
        }
        let Some(pause) = pause_after(sent_ms, unix_ms()) else {
            return;
        };
        let now = tokio::time::Instant::now();
        let was_throttled = extend_deadline(&mut self.deadlines.lock().throttled_until, now, pause);
        // info! not warn!: nothing was refused. The pause keeps the next query from being refused.
        if !was_throttled {
            // as_millis() returns u128; truncation impossible (at most ~61 s here)
            #[allow(clippy::cast_possible_truncation)]
            let pause_ms = pause.as_millis() as u64;
            info!(
                used,
                limit,
                pause_ms,
                "Binance request weight near the per-minute limit, pausing queries until the next minute"
            );
        }
    }

    // no clear() method — cooldowns expire naturally via wait_if_blocked().
    // A previous unconditional clear() on success raced with concurrent calls:
    // call A succeeds → clears cooldown → call B's 429 cooldown is erased.
}

/// Push `deadline` out to `now + delay` unless it is already later, and return whether it was
/// still active at `now`.
fn extend_deadline(
    deadline: &mut Option<tokio::time::Instant>,
    now: tokio::time::Instant,
    delay: Duration,
) -> bool {
    let was_active = deadline.is_some_and(|until| until > now);
    // A wait read from Binance is not bounded, and one past the clock's range would panic on
    // `+`. The longest a REST `Retry-After` can carry, `u32::MAX` seconds (~136 years), stands
    // in for it: the cooldown still outlasts the process.
    let new_deadline = now
        .checked_add(delay)
        .unwrap_or_else(|| now + Duration::from_secs(u64::from(u32::MAX)));
    *deadline = Some(deadline.map_or(new_deadline, |existing| existing.max(new_deadline)));
    was_active
}

/// Whether a reported rate limit is the per-minute request weight, the one the pause follows.
fn is_weight_per_minute(kind: &RateLimitType, interval: &Interval, interval_num: u32) -> bool {
    *kind == RateLimitType::RequestWeight && *interval == Interval::Minute && interval_num == 1
}

/// The current time in Unix milliseconds, the clock a weight observation's send time is taken
/// from.
pub(crate) fn unix_ms() -> u128 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
}

/// How long queries pause, observed at `now_ms`, for a request sent at `sent_ms` (both Unix
/// milliseconds): until just after the UTC minute it was sent in ends, or `None` once it has.
fn pause_after(sent_ms: u128, now_ms: u128) -> Option<Duration> {
    let resume_ms = sent_ms - sent_ms % 60_000 + 60_000 + u128::from(CLOCK_SKEW_SLACK_MS);
    let remaining_ms = resume_ms.checked_sub(now_ms).filter(|&ms| ms > 0)?;
    // At most a minute plus the slack, so it fits in a u64.
    #[allow(clippy::cast_possible_truncation)]
    Some(Duration::from_millis(remaining_ms as u64))
}

/// A request held back unsent: the rate-limit cooldown has more than [`MAX_COOLDOWN_SLEEP`] left.
///
/// [`rest_call_with_retry`] returns it inside its `anyhow::Error`, and
/// [`classify_rest_query_error`] and [`classify_rest_order_error`] report it as
/// [`ApiError::RateLimit`] with `remaining` as the wait.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("Binance rate-limit cooldown active for {remaining:?} more; request not sent")]
pub(crate) struct CooldownActive {
    /// How long the cooldown has left.
    pub(crate) remaining: Duration,
}

/// The wait Binance sent with a REST rate-limit refusal (`Retry-After`, in seconds), as
/// `binance-sdk` reports it on a 429 or 418. `None` for any other error, or a refusal without the
/// header. A zero wait is read as none, so the refusal still starts the default cooldown rather
/// than inviting an immediate retry.
fn rest_retry_after(ce: &ConnectorError) -> Option<Duration> {
    ce.retry_after()
        .filter(|&secs| secs > 0)
        .map(|secs| Duration::from_secs(u64::from(secs)))
}

/// The wait until `until_ms`, the Unix-millisecond time a WS-API refusal's `retryAfter` says a ban
/// or limit lifts, observed at `now_ms` (from [`unix_ms`]): until [`CLOCK_SKEW_SLACK_MS`] after
/// it, or zero once that has passed.
///
/// `until_ms` is on Binance's clock and `now_ms` on this host's. Binance also sends its own time
/// with the refusal, which would cancel the skew, but `binance-sdk` does not expose it. A local
/// clock ahead of Binance's by more than the slack shortens the wait, and a request sent before
/// the ban lifts lengthens it.
fn wait_until_epoch_ms(until_ms: i64, now_ms: u128) -> Duration {
    // A negative time is already past.
    let resume_ms = u128::try_from(until_ms).unwrap_or(0) + u128::from(CLOCK_SKEW_SLACK_MS);
    let wait_ms = resume_ms.saturating_sub(now_ms);
    Duration::from_millis(u64::try_from(wait_ms).unwrap_or(u64::MAX))
}

/// Check if an anyhow::Error from binance-sdk REST is a rate-limit error.
/// Covers HTTP 429 / -1003 (WAF/queue overflow: requests rejected before execution),
/// -1015 (IP rate-limit ban), and a WAF 403 (`ConnectorError::ForbiddenError` without an auth
/// code or wording). All warrant the same backoff response.
pub(crate) fn is_rate_limit_error(e: &anyhow::Error) -> bool {
    // A WAF 403 carries none of the texts below; recognise it by type so it is backed off too.
    if let Some(ConnectorError::ForbiddenError { msg, code }) = e.downcast_ref::<ConnectorError>()
        && is_waf_block(&with_code(msg, *code))
    {
        return true;
    }
    // iterate the error chain and match against the actual Display strings
    // from binance-sdk's TooManyRequestsError and RateLimitBanError variants.
    // Note: cause.to_string() allocates per chain entry — acceptable since this
    // only runs on error paths.
    for cause in e.chain() {
        let msg = cause.to_string();
        if msg.contains("Too many requests")
            || msg.contains("been banned for exceeding rate limits")
            || has_rate_limit_error_code(&msg)
        {
            return true;
        }
    }
    false
}

/// Binance's shortest IP ban: it documents a ban as lasting from 2 minutes to 3 days, longer for
/// repeat offenders. A handshake refused with 418 starts a cooldown this long, since the SDK drops
/// the handshake's `Retry-After`.
pub(crate) const MIN_IP_BAN: Duration = Duration::from_secs(120);

/// Binance refusing a WebSocket handshake for rate limiting.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HandshakeRefusal {
    /// HTTP 429: too many requests.
    Throttled,
    /// HTTP 418: the IP is banned.
    Banned,
}

impl HandshakeRefusal {
    /// The refusal a failed WebSocket connect was, or `None` when Binance did not refuse the
    /// handshake for rate limiting.
    ///
    /// depends on the SDK reporting a refused handshake as `WebsocketError::Handshake` holding
    /// tungstenite's `HTTP error: <status>` text (not a public API contract).
    pub(crate) fn of(e: &anyhow::Error) -> Option<Self> {
        e.chain()
            .find_map(|cause| match cause.downcast_ref::<WebsocketError>()? {
                WebsocketError::Handshake(msg) if msg.contains("HTTP error: 429") => {
                    Some(Self::Throttled)
                }
                WebsocketError::Handshake(msg) if msg.contains("HTTP error: 418") => {
                    Some(Self::Banned)
                }
                _ => None,
            })
    }

    /// The wait the refusal starts a cooldown for. The SDK drops the handshake's `Retry-After`,
    /// so a throttle names none, and the default cooldown applies; a ban lasts at least
    /// [`MIN_IP_BAN`], and connecting sooner would be refused, and could lengthen it.
    pub(crate) fn retry_after(self) -> Option<Duration> {
        match self {
            Self::Throttled => None,
            Self::Banned => Some(MIN_IP_BAN),
        }
    }
}

// ---------------------------------------------------------------------------
// Reconnect budget
// ---------------------------------------------------------------------------

/// Decide, after an account stream's connect failed or its connection ended, whether to try
/// again.
///
/// A rate-limit cooldown longer than a request sleeps through ends the stream at once, rather
/// than spend the budget against a venue that refuses every attempt. Otherwise this spends one
/// attempt and its backoff. A shorter cooldown is left for
/// [`RateLimitTracker::before_stream_connect`] to wait out, which spends no attempt.
///
/// # Errors
/// The reason the stream ends: [`StreamTerminationReason::RateLimited`] for such a cooldown,
/// or [`StreamTerminationReason::ReconnectBudgetExhausted`], with `last_error`, once no
/// attempt is left.
pub(crate) async fn retry_or_end(
    budget: &mut ReconnectBudget,
    rate_limiter: &RateLimitTracker,
    last_error: impl FnOnce() -> String,
) -> Result<(), StreamTerminationReason> {
    if let Some(remaining) = rate_limiter.too_long_to_wait() {
        return Err(end_for_cooldown(remaining));
    }
    if budget.wait().await {
        return Ok(());
    }
    error!(
        attempts = budget.attempts(),
        "Binance account stream: max reconnect attempts exhausted, stream terminating"
    );
    Err(StreamTerminationReason::ReconnectBudgetExhausted {
        attempts: budget.attempts(),
        last_error: last_error(),
    })
}

/// What a reconnect budget spent on dropped connections reports as its last error: a heartbeat
/// timeout, or the venue closing the connection or failing it.
pub(crate) fn disconnect_error(heartbeat_timed_out: bool) -> String {
    if heartbeat_timed_out {
        format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
    } else {
        "WebSocket disconnected (server close/error)".to_string()
    }
}

/// Whether a cooldown with `remaining` left is longer than a request, or a stream's connect, sleeps
/// through: [`MAX_COOLDOWN_SLEEP`].
fn too_long_to_sleep(remaining: Duration) -> bool {
    remaining > MAX_COOLDOWN_SLEEP
}

/// End an account stream for a rate-limit cooldown with `remaining` left.
fn end_for_cooldown(remaining: Duration) -> StreamTerminationReason {
    warn!(
        retry_after_secs = remaining.as_secs(),
        "Binance account stream ending: the rate-limit cooldown outlasts what it waits through"
    );
    StreamTerminationReason::RateLimited {
        retry_after: remaining,
    }
}

// ---------------------------------------------------------------------------
// Parsing helpers (Binance wire strings → rustrade enums)
// ---------------------------------------------------------------------------

pub(crate) fn parse_side(s: &str) -> Option<Side> {
    match s {
        "BUY" => Some(Side::Buy),
        "SELL" => Some(Side::Sell),
        _ => {
            warn!(side = s, "unknown Binance order side");
            None
        }
    }
}

/// What an order row or `executionReport` says about the order's `trailingDelta`.
///
/// Binance sets a trailing order as a conditional type plus a `trailingDelta`, so whether a
/// `STOP_LOSS` is a fixed stop or a trailing one depends on this, not on its type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TrailingDelta {
    /// The source reports trailing deltas and this order has none: it does not trail.
    Absent,
    /// The order trails by this many basis points.
    BasisPoints(u32),
    /// Whether the order trails cannot be told: the source does not report the field (margin's
    /// REST order queries, see #541) or reported it unreadably.
    Unknown,
}

impl TrailingDelta {
    /// Read a `trailingDelta` from a source that reports it: absent or zero means the order does
    /// not trail, and a negative one is unreadable.
    pub(crate) fn from_reported(delta: Option<i64>) -> Self {
        match delta {
            None | Some(0) => Self::Absent,
            Some(bips) => u32::try_from(bips).map_or(Self::Unknown, Self::BasisPoints),
        }
    }
}

/// The [`OrderKind`] a Binance order row or `executionReport` describes, from its `type`,
/// `stopPrice` and `trailingDelta`.
///
/// - `MARKET` is [`OrderKind::Market`]; `LIMIT` and `LIMIT_MAKER` are [`OrderKind::Limit`].
/// - `STOP_LOSS`, `STOP_LOSS_LIMIT`, `TAKE_PROFIT` and `TAKE_PROFIT_LIMIT` without a trailing delta
///   are [`OrderKind::Stop`], [`OrderKind::StopLimit`], [`OrderKind::TakeProfit`] and
///   [`OrderKind::TakeProfitLimit`], triggered at `stopPrice`.
/// - `STOP_LOSS` and `STOP_LOSS_LIMIT` with a trailing delta are an [`OrderKind::TrailingStop`] and
///   an [`OrderKind::TrailingStopFixedLimit`] of that many basis points, whichever offset type
///   they were placed with. A positive `stopPrice` alongside is their `activation_price`.
///
/// Returns `None`, with a warning, for an order this cannot describe faithfully, so that it is
/// dropped and its listing reads as incomplete rather than describing a different order:
/// - a type this version does not know;
/// - a fixed-trigger order without a positive `stopPrice`;
/// - a trailing take-profit, which no [`OrderKind`] describes;
/// - a conditional order whose trailing delta is [`TrailingDelta::Unknown`], which could be either
///   a fixed or a trailing order;
/// - a trailing order whose `stopPrice` does not parse, whose activation price is then unknown.
///
/// The warnings for the trailing take-profit and the unknown trailing delta come from the venue's
/// shape, not from one bad order, and recur on every listing and report of the same live order;
/// each warns on its first occurrence and every 1000th after (see [`sampled`]), and logs the rest
/// at `debug`.
pub(crate) fn decode_order_kind(
    exchange: ExchangeId,
    order_id: Option<i64>,
    order_type: &str,
    stop_price: Option<&str>,
    trailing_delta: TrailingDelta,
) -> Option<OrderKind> {
    let parsed_stop_price = stop_price.map(Decimal::from_str);
    // Not sampled: it signals a malformed venue field, not the venue's shape.
    let unparseable_stop_price = stop_price.filter(|_| matches!(parsed_stop_price, Some(Err(_))));
    let stop_price = parsed_stop_price
        .and_then(Result::ok)
        .filter(|price| price.is_sign_positive() && !price.is_zero());
    let triggered = |kind: fn(Decimal) -> OrderKind| {
        if stop_price.is_none() {
            warn!(%exchange, ?order_id, order_type, "Binance conditional order has no positive stopPrice, dropping it");
        }
        stop_price.map(kind)
    };
    match (order_type, trailing_delta) {
        ("MARKET", _) => Some(OrderKind::Market),
        ("LIMIT" | "LIMIT_MAKER", _) => Some(OrderKind::Limit),
        (_, TrailingDelta::Unknown) if is_conditional_order_type(order_type) => {
            static SEEN: AtomicU64 = AtomicU64::new(0);
            let (count, warns) = sampled(&SEEN);
            if warns {
                warn!(%exchange, ?order_id, order_type, count, "Binance conditional order may be a trailing one and its trailingDelta is unknown, dropping it; further ones are logged at debug, with a warning every 1000th");
            } else {
                debug!(%exchange, ?order_id, order_type, count, "Binance conditional order may be a trailing one and its trailingDelta is unknown, dropping it");
            }
            None
        }
        ("STOP_LOSS", TrailingDelta::Absent) => {
            triggered(|trigger_price| OrderKind::Stop { trigger_price })
        }
        ("STOP_LOSS_LIMIT", TrailingDelta::Absent) => {
            triggered(|trigger_price| OrderKind::StopLimit { trigger_price })
        }
        ("TAKE_PROFIT", TrailingDelta::Absent) => {
            triggered(|trigger_price| OrderKind::TakeProfit { trigger_price })
        }
        ("TAKE_PROFIT_LIMIT", TrailingDelta::Absent) => {
            triggered(|trigger_price| OrderKind::TakeProfitLimit { trigger_price })
        }
        ("STOP_LOSS" | "STOP_LOSS_LIMIT", TrailingDelta::BasisPoints(_))
            if unparseable_stop_price.is_some() =>
        {
            warn!(%exchange, ?order_id, order_type, stop_price = ?unparseable_stop_price, "Binance trailing order's stopPrice, its activation price, does not parse, dropping it");
            None
        }
        ("STOP_LOSS", TrailingDelta::BasisPoints(bips)) => Some(OrderKind::TrailingStop {
            offset: Decimal::from(bips),
            offset_type: TrailingOffsetType::BasisPoints,
            activation_price: stop_price,
        }),
        ("STOP_LOSS_LIMIT", TrailingDelta::BasisPoints(bips)) => {
            Some(OrderKind::TrailingStopFixedLimit {
                offset: Decimal::from(bips),
                offset_type: TrailingOffsetType::BasisPoints,
                activation_price: stop_price,
            })
        }
        ("TAKE_PROFIT" | "TAKE_PROFIT_LIMIT", TrailingDelta::BasisPoints(_)) => {
            static SEEN: AtomicU64 = AtomicU64::new(0);
            let (count, warns) = sampled(&SEEN);
            if warns {
                warn!(%exchange, ?order_id, order_type, count, "Binance trailing order of a type OrderKind cannot describe, dropping it; further ones are logged at debug, with a warning every 1000th");
            } else {
                debug!(%exchange, ?order_id, order_type, count, "Binance trailing order of a type OrderKind cannot describe, dropping it");
            }
            None
        }
        _ => {
            warn!(%exchange, ?order_id, order_type, "unsupported Binance order type");
            None
        }
    }
}

/// Count one more occurrence of a recurring condition on `seen`, a process-wide counter per
/// condition, and say whether it warns: the first does, and every 1000th after it, as in
/// [`log_unrecognised_frame`]. Returns the running count with it.
fn sampled(seen: &AtomicU64) -> (u64, bool) {
    let count = seen.fetch_add(1, Ordering::Relaxed) + 1;
    (count, count == 1 || count.is_multiple_of(1000))
}

/// Whether a Binance order `type` is conditional, and so may carry a `trailingDelta`.
fn is_conditional_order_type(order_type: &str) -> bool {
    matches!(
        order_type,
        "STOP_LOSS" | "STOP_LOSS_LIMIT" | "TAKE_PROFIT" | "TAKE_PROFIT_LIMIT"
    )
}

/// Whether orders of `kind` carry a limit price in Binance's `price` field, which reads zero on
/// every other kind.
fn kind_has_limit_price(kind: &OrderKind) -> bool {
    match kind {
        OrderKind::Limit
        | OrderKind::StopLimit { .. }
        | OrderKind::TakeProfitLimit { .. }
        | OrderKind::TrailingStopLimit { .. }
        | OrderKind::TrailingStopFixedLimit { .. } => true,
        OrderKind::Market
        | OrderKind::Stop { .. }
        | OrderKind::TakeProfit { .. }
        | OrderKind::TrailingStop { .. } => false,
    }
}

pub(crate) fn parse_time_in_force(tif: &str) -> TimeInForce {
    match tif {
        "GTC" => TimeInForce::GoodUntilCancelled { post_only: false },
        "GTX" => TimeInForce::GoodUntilCancelled { post_only: true },
        "IOC" => TimeInForce::ImmediateOrCancel,
        "FOK" => TimeInForce::FillOrKill,
        "GTD" => TimeInForce::GoodUntilEndOfDay,
        _ => {
            warn!(
                time_in_force = tif,
                "unknown Binance TimeInForce, defaulting to GTC"
            );
            TimeInForce::GoodUntilCancelled { post_only: false }
        }
    }
}

// ---------------------------------------------------------------------------
// Order-response conversion (REST open-order / all-order endpoints)
// ---------------------------------------------------------------------------

/// The subset of a Binance order-response struct that [`convert_open_order`] and
/// [`convert_ended_order`] read.
///
/// binance-sdk generates a *distinct* response type per endpoint, with no shared trait:
/// `AllOrdersResponseInner`, `GetOpenOrdersResponseInner` and `GetOrderResponse` on spot,
/// `QueryMarginAccountsOpenOrdersResponseInner` and `QueryMarginAccountsOrderResponse` on margin.
/// Those structs are emphatically not
/// interchangeable -- the spot family carries substantially more fields than the margin one, and
/// several fields share a name while differing in type -- but the fourteen named here are
/// identical in name and type across all of them. The one that is not, `trailingDelta`, the margin
/// structs lack entirely, so each impl says how it reads it.
///
/// Naming that read subset is what makes a single converter safe to share. The dependency surface
/// is explicit, so an SDK change to any *other* field cannot silently alter order parsing, and a
/// correction to the conversion reaches every Binance client at once instead of one of them.
pub(crate) trait BinanceOrderFields {
    fn order_id(&self) -> Option<i64>;
    fn client_order_id(&self) -> Option<&str>;
    fn side(&self) -> Option<&str>;
    fn price(&self) -> Option<&str>;
    fn orig_qty(&self) -> Option<&str>;
    fn executed_qty(&self) -> Option<&str>;
    fn order_type(&self) -> Option<&str>;
    fn time_in_force(&self) -> Option<&str>;
    fn time(&self) -> Option<i64>;
    /// When the order last changed state, as opposed to [`BinanceOrderFields::time`], which is when
    /// it was created. Every endpoint implemented below carries it.
    fn update_time(&self) -> Option<i64>;
    fn symbol(&self) -> Option<&str>;
    /// The order's status. `openOrders` serves only live orders, but `allOrders` also serves
    /// cancelled, expired and filled ones through the same accessors, so [`convert_open_order`]
    /// reads it rather than trusting the endpoint.
    fn status(&self) -> Option<&str>;
    /// The quote quantity the order has traded so far, spelled `cummulativeQuoteQty` by Binance.
    fn cumulative_quote_qty(&self) -> Option<&str>;
    /// `stopPrice`: a conditional order's trigger, or a trailing order's activation price.
    fn stop_price(&self) -> Option<&str>;
    /// `trailingDelta`. The margin order queries do not carry it (#541), so on those every order
    /// reads [`TrailingDelta::Unknown`].
    fn trailing_delta(&self) -> TrailingDelta;
}

/// Implement [`BinanceOrderFields`] for SDK response types that share these field names.
///
/// Every struct listed below declares the fourteen shared fields with the same types, so those
/// accessors are identical; a macro keeps them from drifting apart under hand-editing. Each list
/// names how its structs read `trailingDelta`.
macro_rules! impl_binance_order_fields {
    (reports_trailing_delta: $($t:ty),* $(,)?) => {
        impl_binance_order_fields!(@impl |order| TrailingDelta::from_reported(order.trailing_delta); $($t),*);
    };
    (omits_trailing_delta: $($t:ty),* $(,)?) => {
        impl_binance_order_fields!(@impl |_order| TrailingDelta::Unknown; $($t),*);
    };
    (@impl |$order:ident| $trailing_delta:expr; $($t:ty),*) => {
        $(
            impl BinanceOrderFields for $t {
                fn order_id(&self) -> Option<i64> { self.order_id }
                fn client_order_id(&self) -> Option<&str> { self.client_order_id.as_deref() }
                fn side(&self) -> Option<&str> { self.side.as_deref() }
                fn price(&self) -> Option<&str> { self.price.as_deref() }
                fn orig_qty(&self) -> Option<&str> { self.orig_qty.as_deref() }
                fn executed_qty(&self) -> Option<&str> { self.executed_qty.as_deref() }
                fn order_type(&self) -> Option<&str> { self.r#type.as_deref() }
                fn time_in_force(&self) -> Option<&str> { self.time_in_force.as_deref() }
                fn time(&self) -> Option<i64> { self.time }
                fn update_time(&self) -> Option<i64> { self.update_time }
                fn symbol(&self) -> Option<&str> { self.symbol.as_deref() }
                fn status(&self) -> Option<&str> { self.status.as_deref() }
                fn cumulative_quote_qty(&self) -> Option<&str> {
                    self.cummulative_quote_qty.as_deref()
                }
                fn stop_price(&self) -> Option<&str> { self.stop_price.as_deref() }
                fn trailing_delta(&self) -> TrailingDelta {
                    let $order = self;
                    $trailing_delta
                }
            }
        )*
    };
}

impl_binance_order_fields!(
    reports_trailing_delta: binance_sdk::spot::rest_api::AllOrdersResponseInner,
    binance_sdk::spot::rest_api::GetOpenOrdersResponseInner,
    binance_sdk::spot::rest_api::GetOrderResponse,
);
impl_binance_order_fields!(
    omits_trailing_delta: binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner,
    binance_sdk::margin_trading::rest_api::QueryMarginAccountsOrderResponse,
);

/// Whether a REST order response's status says the order is resting at the exchange, and may
/// therefore become an `Open` order.
///
/// `NEW` and `PARTIALLY_FILLED` are working orders. `PENDING_NEW` is an order-list leg that waits
/// for its working order to fill; it is at the exchange and `openOrders` returns it, so it is live
/// too. Every other status is terminal (`FILLED`, `CANCELED`, `REJECTED`, `EXPIRED`,
/// `EXPIRED_IN_MATCH`), unused (`PENDING_CANCEL`), or unknown to this version.
///
/// This is an allow-list for the same reason [`trade_order_is_live`] is one: an `Open` snapshot of
/// an order that is no longer live resurrects it. A cancelled order that had partly filled would
/// rest in engine state with quantity remaining, and nothing at the exchange would ever fill or
/// cancel it.
fn rest_order_is_open(status: &str) -> bool {
    matches!(status, "NEW" | "PARTIALLY_FILLED" | "PENDING_NEW")
}

/// Convert a Binance open order into rustrade's `Open` state order.
///
/// Returns `None`, with a warning, for an order whose status is not live (see
/// [`rest_order_is_open`]) or is missing, whichever endpoint served it. `openOrders` serves only
/// live orders, so there this never fires in practice. An order that has ended is read with
/// [`convert_ended_order`] instead.
///
/// `exchange` stamps the resulting [`OrderKey`] and every diagnostic below, so one
/// implementation serves each Binance client without a venue name baked into its warnings.
pub(crate) fn convert_open_order<T: BinanceOrderFields>(
    o: &T,
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
    match o.status() {
        Some(status) if rest_order_is_open(status) => {}
        Some(status) => {
            warn!(%exchange, %instrument, order_id = ?o.order_id(), status, "Binance order is not live, not converting it to an open order");
            return None;
        }
        None => {
            warn!(%exchange, %instrument, order_id = ?o.order_id(), "Binance open order missing status");
            return None;
        }
    }
    let row = convert_order_row(o, exchange, instrument, UnknownTrailingDelta::Drop)?;
    Some(row.map_state(|row| {
        // A live order's fill only grows, so an unknown one reads as nothing filled until the
        // account stream reports more.
        Open::new(
            VenueOrderId::Assigned(row.order_id),
            row.time_exchange,
            row.filled_qty.unwrap_or(Decimal::ZERO),
        )
    }))
}

/// How an order that has ended did end, from a REST order row such as `GET /api/v3/order`'s or
/// `GET /sapi/v1/margin/order`'s, under `key`, the key it was asked for.
///
/// `FILLED` becomes [`InactiveOrderState::FullyFilled`], with the average price worked out from
/// `cummulativeQuoteQty`, or none where Binance reports that negative (not available for some
/// historical orders); `CANCELED` becomes [`InactiveOrderState::Cancelled`], `EXPIRED` and
/// `EXPIRED_IN_MATCH` (self-trade prevention) [`InactiveOrderState::Expired`], each with what
/// filled before; and `REJECTED` becomes [`InactiveOrderState::OpenFailed`].
///
/// Returns `None` for an order still live, and, with a warning, for a row whose status is missing
/// or unknown or that cannot be converted. A caller reads `None` as "not ended", so such an order
/// is asked about again later rather than retired on a guess.
///
/// A conditional order whose trailing delta the row cannot tell (margin, #541) is still reported,
/// rather than left unretired: what matters of an order that has ended is that it ended, and an
/// order held as live after it fired is a protective stop that is not there. It is reported with
/// `recorded_kind`, the kind the client holds the order with, when the caller has one; otherwise
/// with its fixed-trigger kind at `stopPrice`, with a warning, since it may have trailed. Without
/// a recorded kind, one without a positive `stopPrice`, which can only be a trailing order, has no
/// kind to report and still reads as not ended.
pub(crate) fn convert_ended_order<T: BinanceOrderFields>(
    o: &T,
    exchange: ExchangeId,
    key: &UnindexedOrderKey,
    recorded_kind: Option<OrderKind>,
) -> Option<UnindexedInactiveOrder> {
    let instrument = &key.instrument;
    let Some(status) = o.status() else {
        warn!(%exchange, %instrument, cid = %key.cid, "Binance order missing status");
        return None;
    };
    if rest_order_is_open(status) {
        return None;
    }
    let row = convert_order_row(
        o,
        exchange,
        instrument,
        UnknownTrailingDelta::Ended {
            recorded: recorded_kind,
        },
    )?;
    let Some(state) = ended_order_state(
        status,
        row.state.order_id.clone(),
        row.state.time_exchange,
        row.quantity,
        row.state.filled_qty,
        |filled_qty| {
            binance_avg_price(
                exchange,
                &row.state.order_id,
                o.cumulative_quote_qty(),
                filled_qty,
            )
        },
    ) else {
        warn!(%exchange, %instrument, cid = %key.cid, status, "Binance order has a status this version does not know, treating it as not ended");
        return None;
    };
    let mut order = row.map_state(|_| state);
    order.key = key.clone();
    Some(order)
}

/// How an order ended, from its Binance `status`, or `None` for a status that does not end it.
///
/// `FILLED` becomes [`InactiveOrderState::FullyFilled`] with the reported fill, or the whole
/// `quantity` when that is unknown, and the average price `avg_price` works out for that fill
/// (asked only then); `CANCELED` becomes [`InactiveOrderState::Cancelled`], `EXPIRED` and
/// `EXPIRED_IN_MATCH` (self-trade prevention) [`InactiveOrderState::Expired`], each with what
/// filled before, `None` when that is unknown; and `REJECTED` becomes
/// [`InactiveOrderState::OpenFailed`]. Shared by the lookup of an order that ended
/// ([`convert_ended_order`]) and the response to placing one ([`placed_order_state`]).
pub(crate) fn ended_order_state<AssetKey, InstrumentKey>(
    status: &str,
    order_id: OrderId,
    time_exchange: DateTime<Utc>,
    quantity: Decimal,
    filled_qty: Option<Decimal>,
    avg_price: impl FnOnce(Decimal) -> Option<Decimal>,
) -> Option<InactiveOrderState<AssetKey, InstrumentKey>> {
    Some(match status {
        // A filled order filled its whole quantity, whether or not `executedQty` said so.
        "FILLED" => {
            let filled_qty = filled_qty.unwrap_or(quantity);
            InactiveOrderState::FullyFilled(Filled::new(
                order_id,
                time_exchange,
                filled_qty,
                avg_price(filled_qty),
            ))
        }
        "CANCELED" => {
            InactiveOrderState::Cancelled(Cancelled::new(order_id, time_exchange, filled_qty))
        }
        "EXPIRED" | "EXPIRED_IN_MATCH" => {
            InactiveOrderState::Expired(Expired::new(order_id, time_exchange, filled_qty))
        }
        "REJECTED" => InactiveOrderState::OpenFailed(OrderError::Rejected(
            ApiError::OrderRejected(format!("Binance rejected order {order_id}")),
        )),
        _ => return None,
    })
}

/// The error for a successful answer to an order or cancel request that cannot be read: one that
/// does not decode, or that lacks the `orderId` the client needs.
///
/// The venue acted on the request, so its outcome is unknown, not refused: it is
/// [`OrderError::Connectivity`], and a caller looks the order up rather than treat it as never
/// placed (or, for a cancel, as still working).
pub(crate) fn unreadable_answer(exchange: ExchangeId, problem: &str) -> UnindexedOrderError {
    error!(%exchange, problem, "Binance answered an order request unreadably - outcome unknown");
    OrderError::Connectivity(ConnectivityError::Socket(format!(
        "{exchange} answered unreadably, outcome unknown: {problem}"
    )))
}

/// The fields of Binance's response to placing an order that say what became of it.
#[derive(Debug, Clone)]
pub(crate) struct PlacementResponse<'a> {
    /// `status`; absent from an `ACK` response.
    pub(crate) status: Option<&'a str>,
    pub(crate) order_id: OrderId,
    pub(crate) time_exchange: DateTime<Utc>,
    /// `executedQty`; absent from an `ACK` response.
    pub(crate) executed_qty: Option<&'a str>,
    /// `cummulativeQuoteQty`.
    pub(crate) cumulative_quote_qty: Option<&'a str>,
}

/// The state that `response`, to placing an order of `quantity`, reports.
///
/// An order that ended in the response itself is reported as it ended (see
/// [`ended_order_state`]): an IOC or FOK order that found no liquidity, or partly filled and
/// expired the rest, is `Expired` with what filled, not `Open`. A live status (`NEW`,
/// `PARTIALLY_FILLED`, `PENDING_NEW`), or none, as in an `ACK` response, reads from what filled:
/// `FullyFilled` once it covers `quantity`, otherwise `Open`. So does a status this version does
/// not know, with a warning, since the account stream reports the order's next state either way.
pub(crate) fn placed_order_state(
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
    quantity: Decimal,
    response: PlacementResponse<'_>,
) -> UnindexedOrderState {
    let PlacementResponse {
        status,
        order_id,
        time_exchange,
        executed_qty,
        cumulative_quote_qty,
    } = response;
    // An `ACK` response reports neither, which is expected, so only that does not warn.
    let filled_qty = match (status, executed_qty) {
        (None, None) => None,
        _ => binance_filled_qty(exchange, &order_id, executed_qty),
    };
    let avg_price =
        |filled_qty| binance_avg_price(exchange, &order_id, cumulative_quote_qty, filled_qty);
    if let Some(status) = status {
        if let Some(ended) = ended_order_state(
            status,
            order_id.clone(),
            time_exchange,
            quantity,
            filled_qty,
            avg_price,
        ) {
            return OrderState::Inactive(ended);
        }
        if !rest_order_is_open(status) {
            warn!(%exchange, %instrument, %order_id, status, "Binance placed an order with a status this version does not know, reading it from what filled");
        }
    }
    match filled_qty {
        Some(filled_qty) if filled_qty >= quantity => {
            let avg_price = avg_price(filled_qty);
            OrderState::fully_filled(Filled::new(order_id, time_exchange, filled_qty, avg_price))
        }
        // A live order's fill only grows, so an unknown one reads as nothing filled until the
        // account stream reports more.
        _ => OrderState::active(Open::new(
            VenueOrderId::Assigned(order_id),
            time_exchange,
            filled_qty.unwrap_or(Decimal::ZERO),
        )),
    }
}

/// The quantity order `order_id` has filled, from Binance's cumulative filled quantity (REST
/// `executedQty`, the execution report's `z`): `None`, with a warning, when it is missing or does
/// not parse, so that an unknown fill is not read as zero.
pub(crate) fn binance_filled_qty(
    exchange: ExchangeId,
    order_id: &OrderId,
    filled_qty: Option<&str>,
) -> Option<Decimal> {
    let Some(raw) = filled_qty else {
        warn!(%exchange, %order_id, "Binance did not report how much the order filled");
        return None;
    };
    match Decimal::from_str(raw) {
        Ok(filled_qty) => Some(filled_qty),
        Err(_) => {
            warn!(%exchange, %order_id, filled_qty = raw, "Binance reported an unparseable filled quantity, treating it as unknown");
            None
        }
    }
}

/// The average price of the fills of order `order_id`: the quote traded (`cummulativeQuoteQty`)
/// over the base traded (`filled_qty`).
///
/// `None` when nothing filled or the quote is missing, when Binance reports it negative (its
/// "not available", for some historical orders), and, with a warning, when it does not parse.
pub(crate) fn binance_avg_price(
    exchange: ExchangeId,
    order_id: &OrderId,
    cumulative_quote_qty: Option<&str>,
    filled_qty: Decimal,
) -> Option<Decimal> {
    if filled_qty.is_zero() {
        return None;
    }
    let quote = cumulative_quote_qty?;
    match Decimal::from_str(quote) {
        Ok(quote) if quote.is_sign_negative() => None,
        Ok(quote) => quote.checked_div(filled_qty),
        Err(_) => {
            warn!(%exchange, %order_id, cummulative_quote_qty = quote, "Binance: failed to parse cummulativeQuoteQty; average price unavailable");
            None
        }
    }
}

/// What a REST order row says about the order's state, whatever its status.
#[derive(Debug)]
struct OrderRow {
    order_id: OrderId,
    /// When the order last changed state.
    time_exchange: DateTime<Utc>,
    /// `None` when Binance did not report it.
    filled_qty: Option<Decimal>,
}

/// How [`convert_order_row`] reads a conditional order whose trailing delta is
/// [`TrailingDelta::Unknown`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum UnknownTrailingDelta {
    /// Drop the row: a live order described wrongly misleads whatever acts on it.
    Drop,
    /// The order has ended (see [`convert_ended_order`]): report it with `recorded`, the kind the
    /// client holds it with, or failing that read it as a fixed-trigger order, since the kind of an
    /// order that has ended no longer steers anything.
    Ended { recorded: Option<OrderKind> },
}

/// Convert the fields of a REST order row that do not depend on its status, shared by
/// [`convert_open_order`] and [`convert_ended_order`]. Returns `None`, with a warning, when a field
/// the order cannot be described without is missing or unparseable.
fn convert_order_row<T: BinanceOrderFields>(
    o: &T,
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
    unknown_trailing_delta: UnknownTrailingDelta,
) -> Option<Order<ExchangeId, InstrumentNameExchange, OrderRow>> {
    let Some(order_id_raw) = o.order_id() else {
        warn!(%exchange, %instrument, "Binance order missing orderId");
        return None;
    };
    let order_id = OrderId(format_smolstr!("{}", order_id_raw));
    if o.client_order_id().is_none() {
        warn!(%exchange, %instrument, order_id = %order_id_raw, "Binance order missing clientOrderId, using orderId as fallback — order may not reconcile with engine state");
    }
    let cid = ClientOrderId::new(
        o.client_order_id()
            .unwrap_or(&format_smolstr!("{}", order_id_raw)),
    );
    let side = match o.side() {
        // parse_side already logs a warning on unknown values
        Some(s) => parse_side(s)?,
        None => {
            warn!(%exchange, %instrument, order_id = %order_id_raw, "Binance order missing side");
            return None;
        }
    };
    let quantity = match o.orig_qty().and_then(|s| Decimal::from_str(s).ok()) {
        Some(v) => v,
        None => {
            warn!(%exchange, %instrument, order_id = %order_id_raw, "Binance order missing/unparseable origQty");
            return None;
        }
    };
    let filled_qty = binance_filled_qty(exchange, &order_id, o.executed_qty());
    let kind = match o.order_type() {
        Some(t) => {
            let trailing_delta = o.trailing_delta();
            // decode_order_kind already logs a warning on what it cannot describe
            let decode = |trailing_delta| {
                decode_order_kind(
                    exchange,
                    Some(order_id_raw),
                    t,
                    o.stop_price(),
                    trailing_delta,
                )
            };
            let undescribed =
                trailing_delta == TrailingDelta::Unknown && is_conditional_order_type(t);
            match unknown_trailing_delta {
                UnknownTrailingDelta::Ended {
                    recorded: Some(recorded),
                } if undescribed => {
                    debug!(%exchange, %instrument, order_id = %order_id_raw, order_type = t, kind = %recorded, "Binance order has ended and its row cannot tell whether it trailed; reporting it with the kind it was held with");
                    recorded
                }
                UnknownTrailingDelta::Ended { recorded: None } if undescribed => {
                    warn!(%exchange, %instrument, order_id = %order_id_raw, order_type = t, "Binance order has ended and may have been a trailing one, which its row cannot tell; reporting it with its fixed trigger");
                    decode(TrailingDelta::Absent)?
                }
                _ => decode(trailing_delta)?,
            }
        }
        None => {
            warn!(%exchange, %instrument, order_id = %order_id_raw, "Binance order missing type");
            return None;
        }
    };
    // Binance reports a price of zero on kinds that carry no limit price.
    let price = if kind_has_limit_price(&kind) {
        o.price().and_then(|s| Decimal::from_str(s).ok())
    } else {
        None
    };
    let time_in_force = parse_time_in_force(o.time_in_force().unwrap_or("GTC"));
    // `update_time` over `time`: `Open::time_exchange` orders an order's states, and the engine
    // discards a snapshot older than the state it already tracks. `time` is the creation stamp and
    // is identical across every snapshot of one order, so a snapshot carrying it is discarded the
    // moment a WebSocket fill has advanced the tracked order past creation -- which is exactly the
    // partially-filled order this fetch exists to reconcile. For an order that has ended it is
    // when it ended.
    let time_exchange = match o
        .update_time()
        .or_else(|| o.time())
        .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
    {
        Some(ts) => ts,
        None => {
            warn!(%exchange, %instrument, order_id = %order_id_raw, "Binance order missing/unparseable time, using now");
            Utc::now()
        }
    };

    Some(Order {
        key: OrderKey::new(
            exchange,
            instrument.clone(),
            // Binance doesn't carry strategy IDs in any response field.
            // Callers must reconcile orders by ClientOrderId or OrderId — never StrategyId.
            StrategyId::unknown(),
            cid,
        ),
        side,
        price,
        quantity,
        kind,
        time_in_force,
        state: OrderRow {
            order_id,
            time_exchange,
            filled_qty,
        },
    })
}

/// One symbol's `openOrders` rows after conversion.
#[derive(Debug)]
pub(crate) struct OpenOrderListing {
    pub(crate) orders: Vec<Order<ExchangeId, InstrumentNameExchange, Open>>,
    /// Whether `orders` shows every row the venue listed under the `clientOrderId` it was placed
    /// with. This is the `orders_complete` an account snapshot built from the listing may claim.
    pub(crate) complete: bool,
}

/// Convert one symbol's `openOrders` rows with [`convert_open_order`], recording whether the
/// result can stand for the whole listing.
///
/// A row the converter drops may be a live order, and a row without a `clientOrderId` is kept under
/// its `orderId`, where the order it was placed as cannot be found. Either leaves the listing
/// unable to say that an order missing from it is gone, so either makes it incomplete. A dropped
/// row that really was finished makes it incomplete too; `openOrders` does not serve those, and
/// erring towards "incomplete" costs only a reconciliation.
pub(crate) fn convert_open_order_listing<T: BinanceOrderFields>(
    rows: &[T],
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
) -> OpenOrderListing {
    let mut complete = true;
    let orders = rows
        .iter()
        .filter_map(|row| {
            let order = convert_open_order(row, exchange, instrument);
            complete &= order.is_some() && row.client_order_id().is_some();
            order
        })
        .collect();
    OpenOrderListing { orders, complete }
}

/// Convert an open-order response whose instrument is recovered from its own `symbol` field,
/// rather than supplied by the caller. Used by the no-symbol "return all" path
/// (each client's no-symbol "return all" fetch), where each order may belong to a different instrument. Drops
/// (with a warning) any order missing `symbol`; otherwise delegates to [`convert_open_order`].
pub(crate) fn convert_open_order_owned_symbol<T: BinanceOrderFields>(
    o: &T,
    exchange: ExchangeId,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
    let instrument = match o.symbol() {
        Some(s) => InstrumentNameExchange::new(s),
        None => {
            warn!(%exchange, "Binance open order missing symbol in return-all query, dropping order");
            return None;
        }
    };
    convert_open_order(o, exchange, &instrument)
}

// ---------------------------------------------------------------------------
// Recovered fills: the order's cumulative filled quantity
// ---------------------------------------------------------------------------

/// Where a `myTrades` walk starts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum MyTradesFrom {
    /// Every execution on the instrument from `start` to `end`, both included, in epoch
    /// milliseconds: a [`FillGap`], or a `fetch_trades` span.
    Span { start: i64, end: i64 },
    /// Every execution of this one order, from its first.
    ///
    /// Queried by `orderId` alone, Binance returns an order's oldest executions first, in
    /// ascending trade id, even when the order has more executions than `limit`; later pages
    /// continue with `fromId`. Observed on the Spot testnet (2026-09-24, a five-execution order
    /// read with `limit=2`); Margin serves the same parameters and is assumed to match.
    Order(i64),
}

/// The fields of a `myTrades` execution that fill recovery reads to rebuild an order's
/// cumulative filled quantity. Spot and margin serve different types that share these fields.
pub(crate) trait BinanceExecutionFields {
    /// The trade id, which Binance assigns in increasing order per symbol.
    fn id(&self) -> Option<i64>;
    fn order_id(&self) -> Option<i64>;
    /// The size of this execution.
    fn qty(&self) -> Option<&str>;
    /// When this execution happened, in epoch milliseconds.
    fn time(&self) -> Option<i64>;
}

macro_rules! impl_binance_execution_fields {
    ($($t:ty),* $(,)?) => {
        $(
            impl BinanceExecutionFields for $t {
                fn id(&self) -> Option<i64> { self.id }
                fn order_id(&self) -> Option<i64> { self.order_id }
                fn qty(&self) -> Option<&str> { self.qty.as_deref() }
                fn time(&self) -> Option<i64> { self.time }
            }
        )*
    };
}

impl_binance_execution_fields!(
    binance_sdk::spot::rest_api::MyTradesResponseInner,
    binance_sdk::margin_trading::rest_api::QueryMarginAccountsTradeListResponseInner,
);

/// How long fill recovery may spend reading recovered orders' executions, from the moment it
/// starts.
///
/// Half of [`FILL_RECOVERY_TIMEOUT_SECS`], which bounds the whole recovery and drops every fill
/// not yet sent when it expires. The lookups only enrich fills; they must never cost one. Past
/// this budget the remaining fills go out without a cumulative, as they did before it existed.
pub(crate) const ORDER_EXECUTIONS_BUDGET: Duration =
    Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS / 2);

/// How many orders' executions one instrument's recovery reads at once.
///
/// Recovery already runs up to eight instruments at once, so this allows 32 lookups in flight.
/// Each is weight 4 or 5 (Spot: `GET /api/v3/order`, `myTrades` with `orderId`) or 10 (Margin)
/// against a limit of several thousand a minute, and [`rest_call_with_retry`] backs off when
/// Binance signals the limit.
pub(crate) const ORDER_EXECUTIONS_IN_FLIGHT: usize = 4;

/// The cumulative filled quantity of each recovered execution's order, as of that execution,
/// keyed by trade id.
///
/// `myTrades` reports executions only, with no cumulative, so a recovered fill cannot advance its
/// order on its own. For each order among `recovered`, this walks that order's executions from its
/// first with `fetch` and sums them in trade-id order. That is the figure the WebSocket reports as
/// `z` on the same execution, so a recovered fill and a live one carry the same
/// [`Trade::order_filled_quantity`]. Up to [`ORDER_EXECUTIONS_IN_FLIGHT`] orders are read at once.
///
/// An order is left out, and its recovered fills keep `None`, when its walk fails, when the walk
/// is unusable (see [`order_running_totals`]), or once `deadline` passes. Each case is logged.
/// `None` is what every recovered fill carried before this existed, and the order's state must
/// then be learned some other way.
pub(crate) async fn recovered_order_totals<T, F, Fut>(
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
    recovered: &[T],
    deadline: tokio::time::Instant,
    fetch: F,
) -> fnv::FnvHashMap<i64, Decimal>
where
    T: BinanceExecutionFields,
    F: Fn(i64) -> Fut,
    Fut: Future<Output = Result<Vec<T>, UnindexedClientError>>,
{
    let mut totals = fnv::FnvHashMap::default();
    for_each_recovered_order(
        RecoveredOrderLookup {
            exchange,
            instrument,
            reading: "orders' executions",
            lost: "cumulative filled quantity",
        },
        recovered,
        deadline,
        fetch,
        |order_id, executions| match order_running_totals(order_id, &executions) {
            Some(running) => totals.extend(running),
            None => warn!(
                %exchange, %instrument, order_id, executions = executions.len(),
                "fill recovery read an order's executions it cannot sum (a missing id or size, \
                 or another order's execution); its recovered fills carry no cumulative filled \
                 quantity"
            ),
        },
    )
    .await;
    totals
}

/// The client order id of each order among `recovered`, keyed by order id.
///
/// `myTrades` carries no client order id, so a recovered fill cannot report the one its order was
/// placed with on its own: this reads each order with `fetch`, `GET /api/v3/order` (weight 4) or
/// `GET /sapi/v1/margin/order` (weight 10). Callers run it after [`recovered_order_totals`],
/// under the same `deadline`, so it uses only the time those walks left: a fill without its
/// cumulative leaves its order behind, while one without its client order id is only harder to
/// attribute. Up to [`ORDER_EXECUTIONS_IN_FLIGHT`] orders are read at once.
///
/// An order is left out, and its recovered fills carry no
/// [`Trade::client_order_id`](crate::trade::Trade::client_order_id), when its lookup fails, when
/// the venue reports no id for it (`fetch` returns `None`), or once `deadline` passes. Failures
/// and the deadline are logged.
pub(crate) async fn recovered_order_cids<T, F, Fut>(
    exchange: ExchangeId,
    instrument: &InstrumentNameExchange,
    recovered: &[T],
    deadline: tokio::time::Instant,
    fetch: F,
) -> fnv::FnvHashMap<i64, ClientOrderId>
where
    T: BinanceExecutionFields,
    F: Fn(i64) -> Fut,
    Fut: Future<Output = Result<Option<ClientOrderId>, UnindexedClientError>>,
{
    let mut cids = fnv::FnvHashMap::default();
    for_each_recovered_order(
        RecoveredOrderLookup {
            exchange,
            instrument,
            reading: "orders",
            lost: "client order id",
        },
        recovered,
        deadline,
        fetch,
        |order_id, cid| {
            if let Some(cid) = cid {
                cids.insert(order_id, cid);
            }
        },
    )
    .await;
    cids
}

/// What a fill-recovery lookup over recovered orders reads, named in what it logs.
struct RecoveredOrderLookup<'a> {
    exchange: ExchangeId,
    instrument: &'a InstrumentNameExchange,
    /// What each lookup reads, such as "orders' executions".
    reading: &'static str,
    /// What a recovered fill goes without when its order's lookup is lost.
    lost: &'static str,
}

/// Look up each order among `recovered` with `fetch`, up to [`ORDER_EXECUTIONS_IN_FLIGHT`] at
/// once, handing each result to `settle`, until every order is read or `deadline` passes.
///
/// A failed lookup, and the orders still unread at the deadline, are logged and skipped: the
/// lookups only enrich recovered fills, so none of them may hold one back.
async fn for_each_recovered_order<T, R, F, Fut>(
    lookup: RecoveredOrderLookup<'_>,
    recovered: &[T],
    deadline: tokio::time::Instant,
    fetch: F,
    mut settle: impl FnMut(i64, R),
) where
    T: BinanceExecutionFields,
    F: Fn(i64) -> Fut,
    Fut: Future<Output = Result<R, UnindexedClientError>>,
{
    use futures::StreamExt as _;

    let RecoveredOrderLookup {
        exchange,
        instrument,
        reading,
        lost,
    } = lookup;
    let mut order_ids: Vec<i64> = recovered.iter().filter_map(|t| t.order_id()).collect();
    order_ids.sort_unstable();
    order_ids.dedup();

    // `timeout_at` polls before it checks, so past the deadline the first poll would still send
    // the first reads only to drop them.
    if tokio::time::Instant::now() >= deadline {
        if !order_ids.is_empty() {
            warn!(
                %exchange, %instrument,
                orders_left = order_ids.len(),
                budget = ?ORDER_EXECUTIONS_BUDGET,
                "fill recovery ran out of time before reading {reading}; the orders' recovered \
                 fills carry no {lost}"
            );
        }
        return;
    }

    let fetch = &fetch;
    let mut lookups = futures::stream::iter(order_ids.iter().copied())
        .map(|order_id| async move { (order_id, fetch(order_id).await) })
        .buffer_unordered(ORDER_EXECUTIONS_IN_FLIGHT);

    let mut settled = 0;
    loop {
        let (order_id, result) = match tokio::time::timeout_at(deadline, lookups.next()).await {
            Ok(Some(lookup)) => lookup,
            Ok(None) => break,
            Err(_elapsed) => {
                warn!(
                    %exchange, %instrument,
                    orders_left = order_ids.len() - settled,
                    budget = ?ORDER_EXECUTIONS_BUDGET,
                    "fill recovery ran out of time reading {reading}; the remaining orders' \
                     recovered fills carry no {lost}"
                );
                break;
            }
        };
        settled += 1;
        match result {
            Ok(value) => settle(order_id, value),
            Err(error) => warn!(
                %exchange, %instrument, order_id, %error,
                "fill recovery could not read {reading} for an order; its recovered fills carry \
                 no {lost}"
            ),
        }
    }
}

/// `(trade id, cumulative filled quantity as of that execution)` for every execution of one
/// order, or `None` if the walk cannot be trusted to sum.
///
/// A running total is only right if every execution is counted exactly once, in order. So the
/// walk is refused whole, rather than summed around a gap, when an execution has no id or no
/// parseable size, or belongs to a different order: the last would mean the venue ignored the
/// `orderId` filter, and summing another order's fills would report a quantity this order never
/// filled. An execution repeated across pages is counted once.
pub(crate) fn order_running_totals<T: BinanceExecutionFields>(
    order_id: i64,
    executions: &[T],
) -> Option<Vec<(i64, Decimal)>> {
    let mut sized = executions
        .iter()
        .map(|execution| {
            if execution.order_id() != Some(order_id) {
                return None;
            }
            let qty = Decimal::from_str(execution.qty()?).ok()?;
            Some((execution.id()?, qty))
        })
        .collect::<Option<Vec<_>>>()?;

    sized.sort_unstable_by_key(|&(id, _)| id);
    sized.dedup_by_key(|&mut (id, _)| id);

    let mut cumulative = Decimal::ZERO;
    Some(
        sized
            .into_iter()
            .map(|(id, qty)| {
                cumulative += qty;
                (id, cumulative)
            })
            .collect(),
    )
}

// ---------------------------------------------------------------------------
// executionReport conversion (WebSocket user-data stream)
// ---------------------------------------------------------------------------

/// A Binance user-data `executionReport`, as both the spot and margin clients read it.
///
/// binance-sdk generates one report type per stream family, each typing every key strictly, and
/// those types change between releases: until 74.1.1 the margin stream's report type declared `d`,
/// `D`, `j`, `J`, `u`, `U`, `a` and `v` strings, where spot's declared them integers and Binance's
/// docs show numbers. 74.1.1 retyped them to integers, in a patch release. Decoding into an SDK
/// type ties every report to the SDK and the venue agreeing on the type of every key, including
/// keys this crate never reads, and a report that fails to decode is logged and dropped, fill
/// included. This type models only the keys the converter reads and ignores every other, so an
/// unread key cannot fail a decode, and the two clients decode through one definition, which the
/// spot testnet exercises for margin too.
///
/// Every field is optional: which keys a report carries depends on its execution type, and each
/// use below says what it does without one.
///
/// Unlike `d`, the keys read as integers, `i`, `T` and `t`, are integers in Binance's docs and in
/// both SDK models, so they are read strictly. A report carrying one of them in another shape
/// still fails to decode whole; the stream handlers log it with an excerpt of the frame.
#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize)]
// Tests build stream frames from reports.
#[cfg_attr(test, derive(serde::Serialize))]
pub(crate) struct ExecutionReport {
    /// `x`: what happened (`NEW`, `TRADE`, `CANCELED`, ...).
    #[serde(rename = "x")]
    pub(crate) execution_type: Option<String>,
    /// `X`: the order's status *after* this execution, which is not the same question as
    /// `execution_type` -- a `TRADE` may leave the order `PARTIALLY_FILLED`, `FILLED`, or already
    /// retired by a report that overtook it.
    #[serde(rename = "X")]
    pub(crate) order_status: Option<String>,
    /// `s`.
    #[serde(rename = "s")]
    pub(crate) symbol: Option<String>,
    /// `S`.
    #[serde(rename = "S")]
    pub(crate) side: Option<String>,
    /// `i`: the venue's order id.
    #[serde(rename = "i")]
    pub(crate) order_id: Option<i64>,
    /// `c`.
    #[serde(rename = "c")]
    pub(crate) client_order_id: Option<String>,
    /// `T`: transaction time, in milliseconds.
    #[serde(rename = "T")]
    pub(crate) transaction_time: Option<i64>,
    /// `t`: the execution's own id, present only on a `TRADE`.
    #[serde(rename = "t")]
    pub(crate) trade_id: Option<i64>,
    /// `L`: the price of this execution alone.
    #[serde(rename = "L")]
    pub(crate) last_executed_price: Option<String>,
    /// `l`: the quantity of this execution alone.
    #[serde(rename = "l")]
    pub(crate) last_executed_quantity: Option<String>,
    /// `n`.
    #[serde(rename = "n")]
    pub(crate) commission_amount: Option<String>,
    /// `N`: the asset the commission was charged in, which need not be either leg of the traded
    /// pair. Absent and `null` both read as `None`.
    #[serde(rename = "N")]
    pub(crate) commission_asset: Option<String>,
    /// `z`: the *order's* cumulative filled quantity as of this execution, as opposed to
    /// `last_executed_quantity`, which is this execution alone.
    #[serde(rename = "z")]
    pub(crate) cumulative_filled_quantity: Option<String>,
    /// `r`.
    #[serde(rename = "r")]
    pub(crate) reject_reason: Option<String>,
    /// `o`.
    #[serde(rename = "o")]
    pub(crate) order_type: Option<String>,
    /// `p`: the order's limit price, `"0"` on order kinds that carry none.
    #[serde(rename = "p")]
    pub(crate) price: Option<String>,
    /// `q`: the order's total quantity, as opposed to any executed portion of it.
    #[serde(rename = "q")]
    pub(crate) order_quantity: Option<String>,
    /// `f`.
    #[serde(rename = "f")]
    pub(crate) time_in_force: Option<String>,
    /// `P`: a conditional order's trigger, or a trailing order's activation price; `"0"` on order
    /// kinds that carry neither.
    #[serde(rename = "P")]
    pub(crate) stop_price: Option<String>,
    /// `d`: the trailing delta in basis points, sent only for a trailing order. Binance's docs show
    /// a number, but the margin SDK model declared a string until 74.1.1, and margin has no
    /// testnet to confirm which the live stream sends, so either is accepted.
    #[serde(rename = "d")]
    pub(crate) trailing_delta: Option<WireInt>,
}

/// An integer that Binance may send as a JSON number or as a numeric string.
///
/// Decoding never fails on the value itself: anything else is kept as [`WireInt::Unreadable`], so
/// that one odd key costs that key alone rather than the report around it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum WireInt {
    Int(i64),
    /// Neither an integer nor a string holding one; carries the value as received, or what kind
    /// of value it was.
    Unreadable(String),
}

impl<'de> Deserialize<'de> for WireInt {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        deserializer.deserialize_any(WireIntVisitor)
    }
}

/// Reads a [`WireInt`] from whatever JSON value arrives, without building a `serde_json::Value`.
struct WireIntVisitor;

impl<'de> Visitor<'de> for WireIntVisitor {
    type Value = WireInt;

    fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("an integer, or a string holding one")
    }

    fn visit_i64<E: de::Error>(self, v: i64) -> Result<WireInt, E> {
        Ok(WireInt::Int(v))
    }

    fn visit_u64<E: de::Error>(self, v: u64) -> Result<WireInt, E> {
        Ok(i64::try_from(v).map_or_else(|_| WireInt::Unreadable(v.to_string()), WireInt::Int))
    }

    fn visit_f64<E: de::Error>(self, v: f64) -> Result<WireInt, E> {
        Ok(WireInt::Unreadable(v.to_string()))
    }

    fn visit_bool<E: de::Error>(self, v: bool) -> Result<WireInt, E> {
        Ok(WireInt::Unreadable(v.to_string()))
    }

    fn visit_str<E: de::Error>(self, v: &str) -> Result<WireInt, E> {
        Ok(v.parse()
            .map_or_else(|_| WireInt::Unreadable(v.to_owned()), WireInt::Int))
    }

    fn visit_unit<E: de::Error>(self) -> Result<WireInt, E> {
        Ok(WireInt::Unreadable("null".to_owned()))
    }

    fn visit_seq<A: de::SeqAccess<'de>>(self, mut seq: A) -> Result<WireInt, A::Error> {
        while seq.next_element::<de::IgnoredAny>()?.is_some() {}
        Ok(WireInt::Unreadable("an array".to_owned()))
    }

    fn visit_map<A: de::MapAccess<'de>>(self, mut map: A) -> Result<WireInt, A::Error> {
        while map
            .next_entry::<de::IgnoredAny, de::IgnoredAny>()?
            .is_some()
        {}
        Ok(WireInt::Unreadable("an object".to_owned()))
    }
}

// Tests build stream frames from reports. An `Int` serializes as the number it was read from, and
// an `Unreadable` as a string of its text, which reads back as `Unreadable` unless it is an integer.
#[cfg(test)]
impl serde::Serialize for WireInt {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        match self {
            Self::Int(int) => serializer.serialize_i64(*int),
            Self::Unreadable(raw) => serializer.serialize_str(raw),
        }
    }
}

/// Whether a `TRADE` report's order status says the order is still live at the exchange, and may
/// therefore be written into engine state as an `Open` snapshot.
///
/// A `TRADE` report carries the order's status (`X`) alongside the execution. Only
/// `PARTIALLY_FILLED` and `FILLED` say the order reached this execution while working; every other
/// status means some other report owns the order's current state. Writing an `Open` snapshot from
/// one of those would resurrect an order the engine has already retired, leaving a resting order
/// that does not exist at the exchange -- a fill that arrives after its order's terminal report is
/// exactly the ordering this guards against.
fn trade_order_is_live(status: &str) -> bool {
    matches!(status, "PARTIALLY_FILLED" | "FILLED")
}

/// Convert a Binance `executionReport` into rustrade account events, appended to `buf`.
///
/// A `TRADE` report genuinely carries two facts -- the execution print (`l`/`L`) and the order's
/// new cumulative filled quantity (`z`) -- so it appends both a `Trade` and an `OrderSnapshot`.
/// Every other execution type appends at most one event, and an unusable report appends none.
///
/// The `Trade` is always first. A fully-filled snapshot retires its order, and routing a fill
/// against an order that has already been retired is a strictly harder problem than routing it
/// against a live one; appending the execution first keeps the easy ordering.
///
/// `exchange` stamps every event and every diagnostic below, so one implementation serves each
/// Binance client without a venue name baked into its warnings.
// Inherent complexity: one arm per Binance execution type (TRADE, NEW, CANCELED, EXPIRED,
// REJECTED, REPLACE), each validating the fields its own variant needs.
#[allow(clippy::cognitive_complexity)]
pub(crate) fn convert_execution_report(
    report: &ExecutionReport,
    exchange: ExchangeId,
    buf: &mut Vec<UnindexedAccountEvent>,
) {
    let Some(exec_type) = report.execution_type.as_deref() else {
        warn!(%exchange, "Binance executionReport missing execution type (x), dropping");
        return;
    };
    let Some(symbol) = report.symbol.as_deref().map(InstrumentNameExchange::new) else {
        warn!(%exchange, "Binance executionReport missing symbol (s), dropping");
        return;
    };
    // Check order_id first -- if it is missing the event is dropped, so avoid constructing cid.
    let Some(order_id_raw) = report.order_id else {
        warn!(%exchange, %symbol, "Binance executionReport missing orderId (i), dropping");
        return;
    };
    let order_id = OrderId(format_smolstr!("{order_id_raw}"));
    let cid = match report.client_order_id.as_deref() {
        Some(c) => ClientOrderId::new(c),
        None => ClientOrderId::new(order_id.0.as_str()),
    };

    let time_exchange = match report
        .transaction_time
        .and_then(|ms| Utc.timestamp_millis_opt(ms).single())
    {
        Some(t) => t,
        None => {
            warn!(%exchange, %symbol, "Binance executionReport missing/unparseable transaction time (T), using now");
            Utc::now()
        }
    };

    match exec_type {
        "NEW" => {
            buf.extend(convert_order_snapshot(
                report,
                exchange,
                symbol,
                cid,
                order_id,
                time_exchange,
            ));
        }
        "TRADE" => {
            // Partial or full fill.
            let Some(trade_id) = report.trade_id else {
                warn!(%exchange, %symbol, "Binance TRADE event missing trade ID (t), dropping");
                return;
            };
            let trade_id = TradeId(format_smolstr!("{trade_id}"));
            // parse_side already logs a warning on unknown values.
            let Some(side) = report.side.as_deref().and_then(parse_side) else {
                warn!(%exchange, %symbol, "Binance TRADE event missing/unknown side (S), dropping");
                return;
            };
            let last_price = match report.last_executed_price.as_deref() {
                Some(s) => match Decimal::from_str(s) {
                    Ok(v) => v,
                    Err(e) => {
                        warn!(%exchange, %symbol, error = %e, raw = s, "Binance TRADE event unparseable last price (L), dropping fill");
                        return;
                    }
                },
                None => {
                    warn!(%exchange, %symbol, "Binance TRADE event missing last price (L), dropping fill");
                    return;
                }
            };
            let last_qty = match report.last_executed_quantity.as_deref() {
                Some(s) => match Decimal::from_str(s) {
                    Ok(v) => v,
                    Err(e) => {
                        warn!(%exchange, %symbol, error = %e, raw = s, "Binance TRADE event unparseable last qty (l), dropping fill");
                        return;
                    }
                },
                None => {
                    warn!(%exchange, %symbol, "Binance TRADE event missing last qty (l), dropping fill");
                    return;
                }
            };
            // Commission parse failure: log and default to 0 rather than dropping the fill.
            let commission = match Decimal::from_str(
                report.commission_amount.as_deref().unwrap_or("0"),
            ) {
                Ok(v) => v,
                Err(e) => {
                    warn!(%exchange, %symbol, error = %e, "Binance TRADE event unparseable commission (n), defaulting to 0");
                    Decimal::ZERO
                }
            };
            // Use the venue's own commission asset (N: e.g. BNB, USDT, BTC). fees_quote is None
            // here; the indexer computes it if the fee is in the quote or base asset. The
            // "UNKNOWN" fallback (rare: the API omits N): the indexer keeps the trade with its
            // fee zeroed, and warns.
            let fee_asset = report
                .commission_asset
                .as_deref()
                .map(AssetNameExchange::from)
                .unwrap_or_else(|| AssetNameExchange::from("UNKNOWN"));
            // `z` is the order's cumulative filled quantity as of this execution, which is what
            // advances the order; `last_qty` above is this execution alone.
            let order_filled_quantity = report
                .cumulative_filled_quantity
                .as_deref()
                .and_then(|s| Decimal::from_str(s).ok());

            let trade = Trade::new(
                trade_id,
                order_id.clone(),
                symbol.clone(),
                StrategyId::unknown(), // Binance doesn't carry strategy IDs
                time_exchange,
                side,
                last_price,
                last_qty,
                order_filled_quantity,
                AssetFees::new(fee_asset, commission, None),
            );
            // Only the order's own `c`: `cid` falls back to the exchange id without one.
            let trade = match report.client_order_id.as_deref() {
                Some(c) if !c.is_empty() => trade.with_client_order_id(cid.clone()),
                _ => trade,
            };
            buf.push(UnindexedAccountEvent::new(
                exchange,
                AccountEventKind::Trade(trade),
            ));

            // The execution alone does not move the order: `filled_quantity` is only ever carried
            // into engine state by an order snapshot, so without this second event a partially
            // filled order reads as having nothing filled until REST reconciliation refreshes it.
            // `convert_order_snapshot` reads field `z` (cumulative filled quantity), which is
            // exactly what a TRADE report carries, so the same builder serves both arms.
            let order_status = report.order_status.as_deref().unwrap_or_default();
            if trade_order_is_live(order_status) {
                buf.extend(convert_order_snapshot(
                    report,
                    exchange,
                    symbol,
                    cid,
                    order_id,
                    time_exchange,
                ));
            } else {
                trace!(
                    %exchange,
                    %symbol,
                    status = order_status,
                    "Binance TRADE for an order the exchange no longer reports as working — \
                     emitting the execution without an order snapshot"
                );
            }
        }
        "CANCELED" | "EXPIRED" | "EXPIRED_IN_MATCH" => {
            buf.push(cancelled_event(
                report,
                exchange,
                symbol,
                cid,
                order_id,
                time_exchange,
            ));
        }
        "REJECTED" => {
            // Rejected by the matching engine after initial acceptance (e.g. insufficient funds
            // discovered post-validation). Mapped to OrderCancelled with an error state so the
            // engine removes this order. The reason (`r`) goes through the same classifier as a
            // synchronous rejection, so one refusal is reported alike whichever path brings it.
            // `r` is a name such as `INSUFFICIENT_FUNDS`, never a numeric code, so only the
            // classifier's text heuristics can match it: a refused borrow never arrives this way.
            let reject_reason = report.reject_reason.as_deref().unwrap_or("unknown");
            warn!(
                %exchange, %symbol, %order_id, reason = reject_reason,
                "Binance order REJECTED by matching engine"
            );
            let error = parse_binance_api_error(reject_reason.to_string(), &symbol);
            let response = UnindexedOrderResponseCancel {
                key: OrderKey::new(exchange, symbol, StrategyId::unknown(), cid),
                state: Err(UnindexedOrderError::Rejected(error)),
            };
            buf.push(UnindexedAccountEvent::new(
                exchange,
                AccountEventKind::OrderCancelled(response),
            ));
        }
        "REPLACE" => {
            // REPLACE is emitted when an order is replaced via the cancel-replace endpoint. The
            // report describes the CANCELLED original order: field `i` is the original order id
            // (already extracted as `order_id` above). The replacement arrives as a subsequent
            // NEW report with its own order id. Emitting OrderCancelled for the original is what
            // removes it from the engine's open-order book.
            buf.push(cancelled_event(
                report,
                exchange,
                symbol,
                cid,
                order_id,
                time_exchange,
            ));
        }
        _ => {
            // PENDING_NEW and PENDING_CANCEL are transient; the terminal state
            // (NEW/CANCELED/...) follows shortly after.
            trace!(%exchange, exec_type, "Binance ignoring execution type");
        }
    }
}

/// Build the `OrderCancelled` event shared by the `CANCELED`/`EXPIRED`/`REPLACE` arms.
///
/// All of them report an order leaving the book with whatever it had filled (`z`) at that point,
/// and differ only in why -- which the caller has already established by matching on `x`.
fn cancelled_event(
    report: &ExecutionReport,
    exchange: ExchangeId,
    symbol: InstrumentNameExchange,
    cid: ClientOrderId,
    order_id: OrderId,
    time_exchange: DateTime<Utc>,
) -> UnindexedAccountEvent {
    let filled_qty = binance_filled_qty(
        exchange,
        &order_id,
        report.cumulative_filled_quantity.as_deref(),
    );
    let response = UnindexedOrderResponseCancel {
        key: OrderKey::new(exchange, symbol, StrategyId::unknown(), cid),
        state: Ok(Cancelled::new(order_id, time_exchange, filled_qty)),
    };
    UnindexedAccountEvent::new(exchange, AccountEventKind::OrderCancelled(response))
}

/// Build the `OrderSnapshot` event for a report that leaves the order resting at the exchange.
///
/// Serves both the `NEW` arm and the live half of the `TRADE` arm: the fields describing the
/// order -- side, kind, price, quantity, TIF and cumulative filled quantity -- are carried
/// identically by both, so one builder covers the acknowledgement and every fill that follows it.
///
/// Returns `None` when a field the snapshot cannot be built without is missing or unparseable;
/// each such case is warned about individually.
fn convert_order_snapshot(
    report: &ExecutionReport,
    exchange: ExchangeId,
    symbol: InstrumentNameExchange,
    cid: ClientOrderId,
    order_id: OrderId,
    time_exchange: DateTime<Utc>,
) -> Option<UnindexedAccountEvent> {
    // parse_side already logs a warning on unknown values.
    let Some(side) = report.side.as_deref().and_then(parse_side) else {
        warn!(%exchange, %symbol, "Binance order report missing/unknown side (S), dropping snapshot");
        return None;
    };
    let trailing_delta = match &report.trailing_delta {
        None => TrailingDelta::Absent,
        Some(WireInt::Int(delta)) => TrailingDelta::from_reported(Some(*delta)),
        Some(WireInt::Unreadable(raw)) => {
            // A shape Binance sends one way recurs on every report of the order, so sample it.
            static SEEN: AtomicU64 = AtomicU64::new(0);
            let (count, warns) = sampled(&SEEN);
            if warns {
                warn!(%exchange, %symbol, %order_id, raw, count, "Binance order report has an unreadable trailing delta (d); further ones are logged at debug, with a warning every 1000th");
            } else {
                debug!(%exchange, %symbol, %order_id, raw, count, "Binance order report has an unreadable trailing delta (d)");
            }
            TrailingDelta::Unknown
        }
    };
    // Without its type the order's kind is unknown, so the snapshot is dropped, as a REST row
    // without one is, rather than read as a `LIMIT`.
    let Some(order_type) = report.order_type.as_deref() else {
        warn!(%exchange, %symbol, %order_id, "Binance order report missing order type (o), dropping snapshot");
        return None;
    };
    // decode_order_kind already logs a warning on what it cannot describe.
    let kind = decode_order_kind(
        exchange,
        report.order_id,
        order_type,
        report.stop_price.as_deref(),
        trailing_delta,
    )?;
    // Binance sends "0" as the price of kinds that carry no limit price, so read it only on the
    // kinds that do.
    let price = if kind_has_limit_price(&kind) {
        match report.price.as_deref() {
            Some(p) => match Decimal::from_str(p) {
                Ok(v) if !v.is_zero() => Some(v),
                Ok(_) => {
                    trace!(%exchange, %symbol, %kind, "Binance order report has zero price (p) on limit-type order, treating as no limit price");
                    None
                }
                Err(e) => {
                    warn!(%exchange, %symbol, price = p, error = %e, "Binance order report unparseable price (p), dropping snapshot");
                    return None;
                }
            },
            None => {
                warn!(%exchange, %symbol, "Binance limit-type order report missing price (p), dropping snapshot");
                return None;
            }
        }
    } else {
        None
    };
    let quantity = match report.order_quantity.as_deref() {
        Some(q) => match Decimal::from_str(q) {
            Ok(v) => v,
            Err(e) => {
                warn!(%exchange, %symbol, qty = q, error = %e, "Binance order report unparseable quantity (q), dropping snapshot");
                return None;
            }
        },
        None => {
            warn!(%exchange, %symbol, "Binance order report missing quantity (q), dropping snapshot");
            return None;
        }
    };
    let time_in_force = parse_time_in_force(report.time_in_force.as_deref().unwrap_or("GTC"));
    // Field `z`: the order's cumulative filled quantity. Usually 0 on a NEW, but read it there
    // too in case of an immediate partial fill on an aggressive order; on a TRADE it is the whole
    // point of the snapshot.
    let filled_qty = report
        .cumulative_filled_quantity
        .as_deref()
        .and_then(|s| Decimal::from_str(s).ok())
        .unwrap_or(Decimal::ZERO);

    let order = Order {
        key: OrderKey::new(
            exchange,
            symbol,
            StrategyId::unknown(), // Binance doesn't carry strategy IDs
            cid,
        ),
        side,
        price,
        quantity,
        kind,
        time_in_force,
        state: OrderState::active(Open::new(
            VenueOrderId::Assigned(order_id),
            time_exchange,
            filled_qty,
        )),
    };

    Some(UnindexedAccountEvent::new(
        exchange,
        AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot::new(
            order,
        )),
    ))
}

// ---------------------------------------------------------------------------
// Error parsing / classification
// ---------------------------------------------------------------------------

/// Case-insensitive substring search that avoids the `to_lowercase` allocation.
fn contains_ignore_case(haystack: &str, needle: &str) -> bool {
    haystack
        .as_bytes()
        .windows(needle.len())
        .any(|w| w.eq_ignore_ascii_case(needle.as_bytes()))
}

/// Returns true if `msg` contains `code` as a standalone numeric token:
/// not immediately preceded or followed by another ASCII digit.
/// Prevents "-2013" from matching "-20130" (suffix guard) or "1-2013" (prefix guard).
///
/// Iterates all occurrences so that if the first match fails a digit-guard check
/// (e.g. `-2013` found inside `-20130`), a later valid occurrence is not missed.
pub(crate) fn contains_error_code(msg: &str, code: &str) -> bool {
    let code_len = code.len();
    let mut start = 0;
    while let Some(rel) = msg[start..].find(code) {
        let pos = start + rel;
        let prefix_ok = pos == 0 || !msg[..pos].ends_with(|c: char| c.is_ascii_digit());
        let suffix_ok = !msg[pos + code_len..].starts_with(|c: char| c.is_ascii_digit());
        if prefix_ok && suffix_ok {
            return true;
        }
        start = pos + 1;
    }
    false
}

/// Whether a Binance error message means the credentials were refused, on any endpoint.
///
/// By code: `-1002` ("You are not authorized to execute this request"), `-2015` ("Invalid
/// API-key, IP, or permissions for action"), `-1022` ("Signature for this request is not valid")
/// or `-2014` ("API-key format invalid"). By wording, for messages that arrive without a code.
fn is_auth_failure(msg: &str) -> bool {
    contains_error_code(msg, "-1002")
        || contains_error_code(msg, "-2015")
        || contains_error_code(msg, "-1022")
        || contains_error_code(msg, "-2014")
        || contains_ignore_case(msg, "invalid api-key")
        || contains_ignore_case(msg, "invalid signature")
        || contains_ignore_case(msg, "signature for this request is not valid")
}

/// Whether a Binance error message carries a code meaning the venue itself failed the request:
/// `-1000` (unknown error), `-1001` (internal error), `-1006` (unexpected response from the
/// message bus), `-1007` (backend timeout) or `-1008` (server overloaded). Binance says to retry
/// these, and for `-1006`/`-1007` states that the execution status is unknown, so an order that
/// fails this way may still have executed.
fn has_venue_failure_code(msg: &str) -> bool {
    ["-1000", "-1001", "-1006", "-1007", "-1008"]
        .iter()
        .any(|code| contains_error_code(msg, code))
}

/// Whether a Binance error message carries a throttling code, on any endpoint: `-1003` (too many
/// requests) or `-1015` (too many orders). Both are transient.
fn has_rate_limit_error_code(msg: &str) -> bool {
    contains_error_code(msg, "-1003") || contains_error_code(msg, "-1015")
}

/// Binance margin codes for a refused borrow, and the [`BorrowRejectReason`] each means.
///
/// The margin client always sends `sideEffectType`, so an order that needs a borrow is refused
/// with one of these. `-3013` (below the minimum borrow) is a malformed order, not a refusal to
/// lend, so it stays [`ApiError::OrderRejected`].
const BORROW_REJECT_CODES: [(&str, BorrowRejectReason); 5] = [
    // EXCEED_MAX_BORROWABLE: "Your borrow amount has exceed maximum borrow amount."
    ("-3006", BorrowRejectReason::BorrowLimitExceeded),
    // "Borrow not allowed."
    ("-3008", BorrowRejectReason::BorrowDisabled),
    // "Borrow is banned for this asset."
    ("-3012", BorrowRejectReason::BorrowDisabled),
    // "Borrow is banned for this account."
    ("-3014", BorrowRejectReason::BorrowDisabled),
    // INSUFFICIENT_INVENTORY: "The system doesn't have enough asset now."
    ("-3045", BorrowRejectReason::InventoryUnavailable),
];

/// The code and [`BorrowRejectReason`] of the borrow refusal a Binance error message carries,
/// by its code ([`BORROW_REJECT_CODES`]).
fn borrow_reject_code(msg: &str) -> Option<(&'static str, BorrowRejectReason)> {
    BORROW_REJECT_CODES
        .iter()
        .copied()
        .find(|(code, _)| contains_error_code(msg, code))
}

/// Parse Binance error strings to rustrade ApiError, in an order's context.
///
/// Order and cancel paths call it through [`parse_binance_order_rejection`], which first catches
/// the venue-failure codes that leave an order's status unknown.
///
/// depends on binance-sdk's internal error formatting (not a public API contract).
/// If the SDK changes its error message format, these string matches may silently stop working.
/// Matches numeric error codes first (stable), then falls back to message text heuristics.
pub(crate) fn parse_binance_api_error(
    error_msg: String,
    instrument: &InstrumentNameExchange,
) -> ApiError<AssetNameExchange, InstrumentNameExchange> {
    // Match on Binance error codes first — these are stable numeric identifiers
    if is_auth_failure(&error_msg) {
        // Auth failures must not be retried as order rejections.
        return ApiError::Unauthenticated(error_msg);
    }
    if has_rate_limit_error_code(&error_msg) {
        return ApiError::rate_limited(None);
    }
    if contains_error_code(&error_msg, "-2011") {
        // -2011: "Unknown order sent" — typically means already cancelled/filled
        return ApiError::OrderAlreadyCancelled;
    }
    if contains_error_code(&error_msg, "-2013") {
        // -2013: "Order does not exist" — on cancel attempts this almost always means
        // the order was already filled or cancelled (a normal race condition).
        return ApiError::OrderAlreadyCancelled;
    }
    if contains_error_code(&error_msg, "-1121") {
        return ApiError::InstrumentInvalid(instrument.clone(), error_msg);
    }
    if let Some((code, reason)) = borrow_reject_code(&error_msg) {
        return ApiError::BorrowRejected(BorrowReject::new(
            reason,
            Some(code.to_owned()),
            error_msg,
        ));
    }
    // -3041 ("Balance is not enough") and -3023 ("You can't transfer out/place order under
    // current margin level.") are the margin account's own collateral running short, not a
    // refused borrow. The wording of -3023 escapes the text heuristic below, so both go by code.
    if contains_error_code(&error_msg, "-3041") || contains_error_code(&error_msg, "-3023") {
        return ApiError::BalanceInsufficient(None, error_msg);
    }
    // -2010 (NEW_ORDER_REJECTED) is deliberately not matched by code: Binance gives it for dozens
    // of reasons ("Account has insufficient balance for requested action.", "Order would trigger
    // immediately.", "Trailing stop orders are not supported for this symbol.", ...), so only the
    // text tells a balance shortfall from the rest.

    // Fall back to case-insensitive message text heuristics (avoid to_lowercase allocation)
    if contains_ignore_case(&error_msg, "duplicate order sent") {
        // -2010's "Duplicate order sent.": the `clOrdId` is already in use. Matched on the whole
        // phrase, since -1102's "Duplicate values for a parameter detected." is a malformed
        // request, not a taken id.
        ApiError::DuplicateClientOrderId(error_msg)
    } else if contains_ignore_case(&error_msg, "insufficient")
        || contains_ignore_case(&error_msg, "not enough")
    {
        // Binance does not name the asset that ran short, so it is left unnamed rather than
        // guessed from the order.
        ApiError::BalanceInsufficient(None, error_msg)
    } else if contains_ignore_case(&error_msg, "rate limit") {
        ApiError::rate_limited(None)
    } else if contains_ignore_case(&error_msg, "unknown order") {
        // -2011/-2013 map to OrderAlreadyCancelled via numeric code above.
        // The text-only fallback here (no code present) maps "unknown order" to
        // OrderRejected — intentionally asymmetric. If the SDK strips error codes,
        // the same semantic maps to a different variant. Acceptable: numeric codes
        // are always present in practice; the text path is a defensive last resort.
        ApiError::OrderRejected(error_msg)
    } else if contains_ignore_case(&error_msg, "invalid symbol") {
        ApiError::InstrumentInvalid(instrument.clone(), error_msg)
    } else {
        ApiError::OrderRejected(error_msg)
    }
}

/// Parse the message of a Binance rejection of an *order* request (place or cancel, REST or WS
/// API) into an [`UnindexedOrderError`].
///
/// A venue failure ([`has_venue_failure_code`]) → [`OrderError::Connectivity`]: the order may
/// or may not have reached the matching engine, and reporting a definitive rejection for one that
/// executed would leave the caller blind to a fill. The venue-failure check comes first on
/// purpose: should a message ever carry both a venue-failure code and an auth or throttle marker,
/// "status unknown" is the label that cannot hide a fill.
///
/// `-3111` (TRAILING_STOP_LIMIT_ONLY, "Trailing stop order only supports limit type") →
/// [`OrderError::UnsupportedOrderType`]: the venue does not take this order type here, so the
/// caller can place another one (a trailing stop-limit) instead. A cross margin account answers
/// a trailing `STOP_LOSS` this way. Binance lists the code, with this one meaning, among its
/// margin (SAPI) errors.
///
/// Everything else is a rejection, mapped by [`parse_binance_api_error`].
pub(crate) fn parse_binance_order_rejection(
    msg: String,
    instrument: &InstrumentNameExchange,
) -> UnindexedOrderError {
    if has_venue_failure_code(&msg) {
        OrderError::Connectivity(ConnectivityError::Socket(msg))
    } else if contains_error_code(&msg, "-3111") {
        OrderError::UnsupportedOrderType(msg)
    } else {
        OrderError::Rejected(parse_binance_api_error(msg, instrument))
    }
}

/// Parse the message of a Binance rejection of a *query* (a non-order request) into a
/// [`UnindexedClientError`], by Binance code.
///
/// Deliberately separate from [`parse_binance_api_error`], which reads codes in an order's
/// context: there `-2011`/`-2013` mean the order was already cancelled, and anything unknown is an
/// [`ApiError::OrderRejected`]. A query has no order to reject, so an unrecognised code is
/// [`ApiError::RequestRejected`].
///
/// - an auth failure ([`is_auth_failure`]) → [`ApiError::Unauthenticated`];
///   `-1003`/`-1015` → [`ApiError::RateLimit`].
/// - a venue failure ([`has_venue_failure_code`]) → [`ConnectivityError::Socket`]: the venue
///   failed the request, not the other way round, and Binance says to retry.
/// - `-1021` (timestamp outside `recvWindow`) → [`ConnectivityError::Socket`]: the request took
///   too long to arrive, or the local clock drifted. The SDK stamps each attempt afresh, so a
///   retry can succeed; a clock that stays wrong keeps failing, so callers should bound retries.
/// - `-1121` → [`ApiError::InstrumentInvalid`] when the request named one instrument; otherwise
///   there is no instrument to attach, so it falls through.
/// - anything else → [`ApiError::RequestRejected`], prefixed with the symbol when the request
///   named one instrument, as Binance's text often does not say which, such as `-11001 Isolated
///   margin account does not exist.`
fn parse_binance_query_rejection(
    msg: String,
    instrument: Option<&InstrumentNameExchange>,
) -> UnindexedClientError {
    let api = if is_auth_failure(&msg) {
        ApiError::Unauthenticated(msg)
    } else if has_rate_limit_error_code(&msg) {
        ApiError::rate_limited(None)
    } else if has_venue_failure_code(&msg) || contains_error_code(&msg, "-1021") {
        return UnindexedClientError::Connectivity(ConnectivityError::Socket(msg));
    } else if let Some(instrument) = instrument.filter(|_| contains_error_code(&msg, "-1121")) {
        ApiError::InstrumentInvalid(instrument.clone(), msg)
    } else if let Some(instrument) = instrument {
        ApiError::RequestRejected(format!("{}: {msg}", instrument.name()))
    } else {
        ApiError::RequestRejected(msg)
    };
    UnindexedClientError::Api(api)
}

/// Classify an `anyhow::Error` from a REST *query* (a non-order request, sent through
/// [`rest_call_with_retry`]) into an [`UnindexedClientError`], so that
/// [`is_transient`](crate::error::ClientError::is_transient) is true only for failures a retry
/// can cure.
///
/// Pass `instrument` when the request named a single instrument, so a `-1121` invalid-symbol
/// rejection can say which.
///
/// - [`CooldownActive`] (held back unsent) → [`ApiError::RateLimit`], with the cooldown's time
///   left as the wait.
/// - The SDK's [`ParamBuildError`] (a request builder missing a required field) is a bug in this
///   client → [`UnindexedClientError::Internal`].
/// - A [`ConnectorError`] is classified by [`classify_connector_error`], and a venue rejection by
///   [`parse_binance_query_rejection`].
/// - Anything else failed before the request was sent: the SDK returns every network and venue
///   failure as a `ConnectorError`, so untyped text comes from its request plumbing (joining or
///   parsing the URL, signing, header values), which fails identically on retry. An auth failure
///   ([`is_auth_failure`]) → [`ApiError::Unauthenticated`], otherwise
///   [`UnindexedClientError::Internal`].
///
/// A response body that arrives but does not decode is a separate path: see
/// [`response_decode_error`].
pub(crate) fn classify_rest_query_error(
    e: &anyhow::Error,
    instrument: Option<&InstrumentNameExchange>,
) -> UnindexedClientError {
    if let Some(cooldown) = e.downcast_ref::<CooldownActive>() {
        return UnindexedClientError::Api(ApiError::rate_limited(Some(cooldown.remaining)));
    }
    if let Some(build) = e.downcast_ref::<ParamBuildError>() {
        return UnindexedClientError::Internal(format!("building Binance request: {build}"));
    }
    if let Some(ce) = e.downcast_ref::<ConnectorError>() {
        return match classify_connector_error(ce) {
            RequestFailure::RateLimited(retry_after) => {
                UnindexedClientError::Api(ApiError::rate_limited(retry_after))
            }
            RequestFailure::Unauthenticated(msg) => {
                UnindexedClientError::Api(ApiError::Unauthenticated(msg))
            }
            RequestFailure::Transport(msg) => {
                UnindexedClientError::Connectivity(ConnectivityError::Socket(msg))
            }
            RequestFailure::Rejected(msg) => parse_binance_query_rejection(msg, instrument),
        };
    }

    let msg = format!("{e:#}");
    if is_auth_failure(&msg) {
        return UnindexedClientError::Api(ApiError::Unauthenticated(msg));
    }

    UnindexedClientError::Internal(msg)
}

/// Whether a REST query for one order failed because Binance does not know the order under the
/// symbol it named: `-2013` (the order does not exist) or `-1121` (the symbol does not). Either way
/// the venue has answered, and its answer is that there is no such order.
pub(crate) fn is_unknown_order(e: &anyhow::Error) -> bool {
    matches!(
        e.downcast_ref::<ConnectorError>(),
        Some(
            ConnectorError::BadRequestError {
                code: Some(-2013 | -1121),
                ..
            } | ConnectorError::NotFoundError {
                code: Some(-2013 | -1121),
                ..
            } | ConnectorError::ConnectorClientError {
                code: Some(-2013 | -1121),
                ..
            }
        )
    )
}

/// Map a failed `RestApiResponse::data()` into [`UnindexedClientError::Internal`].
///
/// In binance-sdk `data()` only runs `serde_json::from_str` over a body already read and
/// decoded, so a failure is a mismatch between a 2xx body and the SDK's model: in practice
/// deterministic, since retrying returns the same body. A body that could not be read or decoded in the
/// first place fails earlier, inside the request, and [`classify_rest_query_error`] reports it as
/// connectivity.
pub(crate) fn response_decode_error(e: ConnectorError) -> UnindexedClientError {
    UnindexedClientError::Internal(format!("decoding Binance response: {e}"))
}

// ---------------------------------------------------------------------------
// REST call retry wrapper
// ---------------------------------------------------------------------------

/// Execute a REST call with rate-limit awareness and retry.
///
/// Waits first for the rate-limit waits `kind` honours (see [`RateLimitTracker`]), and records
/// the weight a successful response reports as used, which can pause later queries.
///
/// A rate-limit refusal ([`is_rate_limit_error`]) is retried in place, up to
/// [`MAX_RATE_LIMIT_RETRIES`] times, after the longer of an exponential backoff (1 s, 2 s, 4 s)
/// and the wait Binance sent, so long as that is at most [`MAX_COOLDOWN_SLEEP`]. A 418 ban is
/// never retried in place: a request sent inside a ban lengthens it. Every refusal pauses all
/// requests on the shared [`RateLimitTracker`]; one returned to the caller pauses them for the
/// wait Binance sent, or [`DEFAULT_RATE_LIMIT_DELAY_SECS`] without one.
///
/// # Errors
///
/// The SDK's error from the last attempt, or [`CooldownActive`] when a cooldown too long to sleep
/// through held the request back unsent. Map either with [`classify_rest_query_error`] or
/// [`classify_rest_order_error`].
///
/// Generic over the SDK `RestApi` type (`R`) so it serves both the spot
/// (`binance_sdk::spot::rest_api::RestApi`) and margin
/// (`binance_sdk::margin_trading::rest_api::RestApi`) clients — the helper never touches
/// `R` itself, it only hands an `Arc<R>` clone to the per-attempt closure. Also usable
/// from concurrent per-instrument futures that hold only `Arc<R>` + `Arc<RateLimitTracker>`.
pub(crate) async fn rest_call_with_retry<R, D>(
    rest: &Arc<R>,
    rate_limiter: &RateLimitTracker,
    kind: RequestKind,
    mut make_call: impl FnMut(
        Arc<R>,
    ) -> Pin<
        Box<dyn std::future::Future<Output = anyhow::Result<RestApiResponse<D>>> + Send>,
    >,
) -> anyhow::Result<RestApiResponse<D>>
where
    // `Arc<R>` is moved into the `+ Send` future, which requires `R: Send + Sync`. Both SDK
    // RestApi types satisfy this; stating it here surfaces the constraint at the definition
    // rather than as a confusing error at the call sites' `Box::pin(async move ...)`.
    R: Send + Sync,
{
    // On the last iteration (attempt == MAX_RATE_LIMIT_RETRIES) a rate-limit error is not
    // retried, so every match arm returns. The post-loop unreachable!() is a runtime safety
    // net — the loop body always returns before exhaustion.
    for attempt in 0..=MAX_RATE_LIMIT_RETRIES {
        rate_limiter.wait_if_blocked(kind).await?;
        let sent_ms = unix_ms();
        match make_call(Arc::clone(rest)).await {
            Ok(response) => {
                rate_limiter.observe_rest(&response, sent_ms);
                return Ok(response);
            }
            Err(e) if is_rate_limit_error(&e) => {
                let refusal = e.downcast_ref::<ConnectorError>();
                let venue_wait = refusal.and_then(rest_retry_after);
                let banned = matches!(refusal, Some(ConnectorError::RateLimitBanError { .. }));
                // An exponential backoff starting at 1 s recovers quickly from a transient burst,
                // but never sooner than Binance asked.
                let backoff = Duration::from_secs(2u64.saturating_pow(attempt).min(30));
                let delay = venue_wait.map_or(backoff, |wait| wait.max(backoff));
                if attempt < MAX_RATE_LIMIT_RETRIES && !banned && delay <= MAX_COOLDOWN_SLEEP {
                    warn!(
                        attempt = attempt.saturating_add(1),
                        max = MAX_RATE_LIMIT_RETRIES,
                        delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
                        "Binance REST rate-limited, retrying"
                    );
                    rate_limiter.on_rate_limited(Some(delay));
                } else {
                    warn!(
                        attempt = attempt.saturating_add(1),
                        banned,
                        venue_wait_ms = venue_wait
                            .map(|wait| u64::try_from(wait.as_millis()).unwrap_or(u64::MAX)),
                        "Binance REST rate-limited, not retrying"
                    );
                    rate_limiter.on_rate_limited(venue_wait);
                    return Err(e);
                }
            }
            Err(e) => return Err(e),
        }
    }
    unreachable!("Binance REST retries exhausted: loop invariant violated")
}

// ---------------------------------------------------------------------------
// Order-kind / TIF classification (rustrade enums → Binance semantics)
// ---------------------------------------------------------------------------

/// Venue-neutral Binance order type — the single source of truth for mapping a rustrade
/// [`OrderKind`] to Binance order semantics, shared by the spot and margin clients.
///
/// Each client maps this onto its own SDK's per-endpoint order-type enum (spot's WS-API
/// `OrderPlaceTypeEnum`, margin's REST `MarginAccountNewOrderTypeEnum`), so neither ever builds
/// the wire string by hand. Keeping the decision logic here (in [`classify_order_kind_tif`])
/// avoids duplicating the match arms across the two clients, which differ only in their SDK
/// output types.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BinanceOrderType {
    Market,
    Limit,
    LimitMaker,
    StopLoss,
    StopLossLimit,
    TakeProfit,
    TakeProfitLimit,
}

/// Venue-neutral Binance time-in-force. Both spot and margin expose exactly `GTC`/`IOC`/`FOK`
/// on their order endpoints; post-only is modelled as [`BinanceOrderType::LimitMaker`] (no TIF),
/// matching Binance's own API.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum BinanceTimeInForce {
    Gtc,
    Ioc,
    Fok,
}

/// Map a rustrade [`OrderKind`] + [`TimeInForce`] to Binance order semantics.
///
/// Returns `None` for combinations Binance does not support (so callers surface
/// `UnsupportedOrderType`). `TrailingStop` classifies to [`BinanceOrderType::StopLoss`] and
/// `TrailingStopFixedLimit` to [`BinanceOrderType::StopLossLimit`], to which the caller adds a
/// `trailingDelta` from [`trailing_delta_basis_points`] and an activation price as `stopPrice`.
/// `TrailingStopLimit` classifies to `None`: Binance's trailing `STOP_LOSS_LIMIT` keeps its limit
/// price fixed while the stop trails, where `TrailingStopLimit`'s limit follows the stop.
pub(crate) fn classify_order_kind_tif(
    kind: OrderKind,
    tif: TimeInForce,
) -> Option<(BinanceOrderType, Option<BinanceTimeInForce>)> {
    match kind {
        OrderKind::Market => Some((BinanceOrderType::Market, None)),
        OrderKind::Limit => match tif {
            TimeInForce::GoodUntilCancelled { post_only: false } => {
                Some((BinanceOrderType::Limit, Some(BinanceTimeInForce::Gtc)))
            }
            TimeInForce::GoodUntilCancelled { post_only: true } => {
                // LIMIT_MAKER is Binance's post-only order type (rejects if
                // it would immediately match as taker)
                Some((BinanceOrderType::LimitMaker, None))
            }
            TimeInForce::FillOrKill => {
                Some((BinanceOrderType::Limit, Some(BinanceTimeInForce::Fok)))
            }
            TimeInForce::ImmediateOrCancel => {
                Some((BinanceOrderType::Limit, Some(BinanceTimeInForce::Ioc)))
            }
            // Binance does not support GTD (good-til-end-of-day), GTC-until-date, MOO, or MOC.
            // Surface as unsupported rather than silently coercing — these have venue-specific
            // semantics (e.g. an end-of-day auto-cancel) that a GTC coercion would silently drop,
            // risking an unintended resting/overnight order.
            TimeInForce::GoodUntilEndOfDay
            | TimeInForce::GoodTillDate { .. }
            | TimeInForce::AtOpen
            | TimeInForce::AtClose => {
                warn!(time_in_force = ?tif, "Binance does not support this TimeInForce");
                None
            }
        },
        // Conditional orders: stop_price/trailing_delta set separately by the caller.
        OrderKind::Stop { .. } => Some((BinanceOrderType::StopLoss, None)),
        OrderKind::TrailingStopFixedLimit {
            offset_type: TrailingOffsetType::Absolute,
            ..
        } => {
            warn!("{ABSOLUTE_TRAILING_OFFSET_UNSUPPORTED}");
            None
        }
        // A fixed-limit trailing stop is a STOP_LOSS_LIMIT with a trailingDelta, so it takes the
        // time in force a stop-limit order does.
        OrderKind::StopLimit { .. } | OrderKind::TrailingStopFixedLimit { .. } => match tif {
            // StopLimit requires TIF like regular Limit orders.
            TimeInForce::GoodUntilCancelled { post_only: false } => Some((
                BinanceOrderType::StopLossLimit,
                Some(BinanceTimeInForce::Gtc),
            )),
            TimeInForce::FillOrKill => Some((
                BinanceOrderType::StopLossLimit,
                Some(BinanceTimeInForce::Fok),
            )),
            TimeInForce::ImmediateOrCancel => Some((
                BinanceOrderType::StopLossLimit,
                Some(BinanceTimeInForce::Ioc),
            )),
            _ => {
                warn!(time_in_force = ?tif, "Binance StopLimit does not support this TimeInForce");
                None
            }
        },
        OrderKind::TakeProfit { .. } => Some((BinanceOrderType::TakeProfit, None)),
        OrderKind::TakeProfitLimit { .. } => match tif {
            // TakeProfitLimit requires TIF like regular Limit orders.
            TimeInForce::GoodUntilCancelled { post_only: false } => Some((
                BinanceOrderType::TakeProfitLimit,
                Some(BinanceTimeInForce::Gtc),
            )),
            TimeInForce::FillOrKill => Some((
                BinanceOrderType::TakeProfitLimit,
                Some(BinanceTimeInForce::Fok),
            )),
            TimeInForce::ImmediateOrCancel => Some((
                BinanceOrderType::TakeProfitLimit,
                Some(BinanceTimeInForce::Ioc),
            )),
            _ => {
                warn!(time_in_force = ?tif, "Binance TakeProfitLimit does not support this TimeInForce");
                None
            }
        },
        // TrailingStop: Binance uses STOP_LOSS with a trailingDelta parameter. Only
        // BasisPoints and Percentage are supported; Absolute requires manual conversion by
        // the caller: basis_points = (absolute / price) * 10000.
        OrderKind::TrailingStop { offset_type, .. } => match offset_type {
            TrailingOffsetType::BasisPoints | TrailingOffsetType::Percentage => {
                Some((BinanceOrderType::StopLoss, None))
            }
            TrailingOffsetType::Absolute => {
                warn!("{ABSOLUTE_TRAILING_OFFSET_UNSUPPORTED}");
                None
            }
        },
        // Binance's trailing STOP_LOSS_LIMIT has a fixed limit price, not one that follows the
        // stop as TrailingStopLimit's does: that is TrailingStopFixedLimit.
        OrderKind::TrailingStopLimit { .. } => {
            warn!("Binance does not support TrailingStopLimit orders");
            None
        }
    }
}

/// Why an Absolute trailing offset is refused, and what to send instead.
const ABSOLUTE_TRAILING_OFFSET_UNSUPPORTED: &str = "Binance TrailingStop does not support an \
     Absolute offset; convert it to basis points: (absolute / price) * 10000";

/// Why a [`OrderKind::TrailingStop`] offset cannot be sent as Binance's `trailingDelta`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TrailingDeltaError {
    /// Binance trails by a share of the price only, so an [`TrailingOffsetType::Absolute`] offset
    /// has no `trailingDelta`.
    AbsoluteOffset,
    /// The offset is not a positive whole number of basis points, or too large to send.
    Invalid(PrecisionViolation),
}

impl TrailingDeltaError {
    /// The order error this refusal is reported as.
    pub(crate) fn into_order_error<AssetKey, InstrumentKey>(
        self,
    ) -> OrderError<AssetKey, InstrumentKey> {
        match self {
            Self::AbsoluteOffset => {
                OrderError::UnsupportedOrderType(ABSOLUTE_TRAILING_OFFSET_UNSUPPORTED.to_owned())
            }
            Self::Invalid(violation) => OrderError::InvalidPrecision(violation),
        }
    }
}

/// Binance's `trailingDelta` for a [`OrderKind::TrailingStop`] `offset`, in basis points.
///
/// A `BasisPoints` offset is sent as it is and a `Percentage` one times 100. Binance takes a whole
/// number of basis points, and rounding would trail by a different distance than asked, so an
/// offset that does not come to one is refused (a `Percentage` offset with more than two decimal
/// places, a fractional `BasisPoints` one), as is one that is not positive or too large to send.
/// Binance enforces its own range on top (the symbol's `TRAILING_DELTA` filter).
pub(crate) fn trailing_delta_basis_points(
    offset: Decimal,
    offset_type: TrailingOffsetType,
) -> Result<i32, TrailingDeltaError> {
    let invalid = |limit| {
        TrailingDeltaError::Invalid(PrecisionViolation::new(
            OrderField::TrailingOffset,
            offset,
            limit,
        ))
    };
    let (basis_points, max_decimal_places) = match offset_type {
        TrailingOffsetType::BasisPoints => (Some(offset), 0),
        TrailingOffsetType::Percentage => (offset.checked_mul(Decimal::ONE_HUNDRED), 2),
        TrailingOffsetType::Absolute => return Err(TrailingDeltaError::AbsoluteOffset),
    };
    if offset.is_sign_negative() || offset.is_zero() {
        return Err(invalid(PrecisionLimit::NotPositive));
    }
    let Some(basis_points) = basis_points else {
        return Err(invalid(PrecisionLimit::NotRepresentable));
    };
    if !basis_points.fract().is_zero() {
        return Err(invalid(PrecisionLimit::DecimalPlaces {
            max: max_decimal_places,
        }));
    }
    i32::try_from(basis_points).map_err(|_| invalid(PrecisionLimit::NotRepresentable))
}

/// What a Binance trailing order sends besides its type: its `trailingDelta`, the fixed limit
/// `price` of a [`OrderKind::TrailingStopFixedLimit`], and an activation price as `stopPrice`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct TrailingParams {
    pub trailing_delta: i32,
    pub price: Option<Decimal>,
    pub stop_price: Option<Decimal>,
}

/// Why a trailing order cannot be sent to Binance as asked.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum TrailingOrderError {
    /// Its offset cannot be sent as `trailingDelta`.
    Delta(TrailingDeltaError),
    /// Its activation price is not positive. Binance reads a zero `stopPrice` as none.
    NonPositiveActivation(Decimal),
    /// A [`OrderKind::TrailingStopFixedLimit`] without the fixed limit `price`, or with one that
    /// is not positive.
    LimitPrice(Option<Decimal>),
}

impl TrailingOrderError {
    /// The order error this refusal is reported as.
    pub(crate) fn into_order_error<AssetKey, InstrumentKey>(
        self,
    ) -> OrderError<AssetKey, InstrumentKey> {
        match self {
            Self::Delta(refused) => refused.into_order_error(),
            Self::NonPositiveActivation(price) => OrderError::Rejected(ApiError::OrderRejected(
                format!("a trailing stop's activation_price must be positive, got {price}"),
            )),
            Self::LimitPrice(price) => OrderError::Rejected(ApiError::OrderRejected(format!(
                "TrailingStopFixedLimit requires a positive price, its fixed limit, got {price:?}"
            ))),
        }
    }
}

/// The trailing fields of `kind`, or `None` when it is not a trailing order Binance places.
/// `price` is the request's limit price. A bad offset is reported before a bad price.
pub(crate) fn trailing_params(
    kind: &OrderKind,
    price: Option<Decimal>,
) -> Result<Option<TrailingParams>, TrailingOrderError> {
    let positive = |value: &Decimal| value.is_sign_positive() && !value.is_zero();
    let (offset, offset_type, activation_price, fixed_limit) = match *kind {
        OrderKind::TrailingStop {
            offset,
            offset_type,
            activation_price,
        } => (offset, offset_type, activation_price, false),
        OrderKind::TrailingStopFixedLimit {
            offset,
            offset_type,
            activation_price,
        } => (offset, offset_type, activation_price, true),
        OrderKind::Market
        | OrderKind::Limit
        | OrderKind::Stop { .. }
        | OrderKind::StopLimit { .. }
        | OrderKind::TakeProfit { .. }
        | OrderKind::TakeProfitLimit { .. }
        | OrderKind::TrailingStopLimit { .. } => return Ok(None),
    };
    let trailing_delta =
        trailing_delta_basis_points(offset, offset_type).map_err(TrailingOrderError::Delta)?;
    if let Some(activation) = activation_price.filter(|activation| !positive(activation)) {
        return Err(TrailingOrderError::NonPositiveActivation(activation));
    }
    let price = if fixed_limit {
        match price.filter(positive) {
            Some(limit) => Some(limit),
            None => return Err(TrailingOrderError::LimitPrice(price)),
        }
    } else {
        None
    };
    Ok(Some(TrailingParams {
        trailing_delta,
        price,
        stop_price: activation_price,
    }))
}

// ---------------------------------------------------------------------------
// Request-error classification (REST and WS API)
// ---------------------------------------------------------------------------

/// How a binance-sdk request failed — a REST [`ConnectorError`] or a WS-API
/// [`WebsocketError::ResponseError`] — before the caller's context (an order or a query) decides
/// which rustrade error that is.
#[derive(Debug, Clone, PartialEq, Eq)]
enum RequestFailure {
    /// HTTP 429 or 418 (throttled or IP-banned), a WS-API `-1003`/`-1015`, or a 403 that is not
    /// an auth failure. Carries the wait Binance sent, measured from when it was read.
    RateLimited(Option<Duration>),
    /// HTTP 401, or a 403 that is an auth failure. Carries the message with the Binance code
    /// spliced in.
    Unauthenticated(String),
    /// The request never completed, the venue failed it (5xx), or a response body could not be
    /// read. Whether the venue acted on it is unknown.
    Transport(String),
    /// The venue answered and refused the request. Carries the message with the Binance code
    /// spliced in, for a code-first parser.
    Rejected(String),
}

/// Splice a Binance code back into a message whose `Display` omits it.
fn with_code(msg: &str, code: Option<i64>) -> String {
    code.map_or_else(|| msg.to_owned(), |c| format!("{c} {msg}"))
}

/// Whether an HTTP 403 is Binance's web application firewall (WAF) limit rather than an auth
/// failure. Binance documents 403 as the WAF limit being violated, and a CDN block answers 403
/// too; it is an auth failure only when its code or wording says so. `msg` carries the code.
fn is_waf_block(msg: &str) -> bool {
    !is_auth_failure(msg)
}

/// Classify an HTTP 403 (REST or WS API). A WAF block was refused at the edge, so an order never
/// reached the matching engine; it is a limit to back off from, which [`is_rate_limit_error`]
/// also recognises so [`rest_call_with_retry`] backs off in place.
fn classify_forbidden(msg: String) -> RequestFailure {
    if is_waf_block(&msg) {
        // `RateLimited` carries no message, so log the body it replaces. A CDN's 403 body is a
        // full HTML page: log only its start. A 403 carries no wait.
        const LOG_EXCERPT_CHARS: usize = 256;
        let excerpt = msg
            .char_indices()
            .nth(LOG_EXCERPT_CHARS)
            .map_or(msg.as_str(), |(end, _)| &msg[..end]);
        warn!(body = %excerpt, "Binance answered 403 (WAF limit); reporting RateLimit");
        RequestFailure::RateLimited(None)
    } else {
        RequestFailure::Unauthenticated(msg)
    }
}

/// Classify a binance-sdk WS-API [`WebsocketError::ResponseError`]: the venue answered a request
/// with a status of 400 or more. `msg` is the error's `Display`, which already carries `code`.
/// `retry_after` is the wait its `retryAfter` asked for (see [`wait_until_epoch_ms`]).
///
/// A negative `code` is Binance's own error code: `-1003`/`-1015` → rate-limited, which is how
/// Binance reports a throttle or a ban on the WS API; anything else goes to a code-first parser.
/// When the response had no error object, the SDK falls back to the HTTP status as `code` (and
/// `"Unknown error"` as the message), so a non-negative `code` is read as a status, as the REST
/// path reads one: 429/418 → rate-limited, 401 → unauthenticated, 403 → [`classify_forbidden`],
/// 5xx → transport (the venue failed it; the status of any order is unknown).
fn classify_ws_response_error(
    code: i64,
    retry_after: Option<Duration>,
    msg: String,
) -> RequestFailure {
    match code {
        429 | 418 | -1003 | -1015 => RequestFailure::RateLimited(retry_after),
        401 => RequestFailure::Unauthenticated(msg),
        403 => classify_forbidden(msg),
        500..=599 => RequestFailure::Transport(msg),
        _ => RequestFailure::Rejected(msg),
    }
}

/// Classify a binance-sdk REST [`ConnectorError`], shared by [`classify_rest_order_error`] and
/// [`classify_rest_query_error`].
///
/// `ConnectorError`'s `Display` **omits** the Binance numeric code (it lives in a separate `code`
/// field), so the code is spliced back into the carried message wherever the caller parses it.
fn classify_connector_error(ce: &ConnectorError) -> RequestFailure {
    match ce {
        ConnectorError::TooManyRequestsError { .. } | ConnectorError::RateLimitBanError { .. } => {
            RequestFailure::RateLimited(rest_retry_after(ce))
        }
        // Splice the code in so callers can tell auth-failure subtypes apart (e.g. -2014
        // invalid key vs -2015 IP/permission).
        ConnectorError::UnauthorizedError { msg, code } => {
            RequestFailure::Unauthenticated(with_code(msg, *code))
        }
        ConnectorError::ForbiddenError { msg, code } => classify_forbidden(with_code(msg, *code)),
        ConnectorError::ServerError { msg, .. } | ConnectorError::NetworkError(msg) => {
            RequestFailure::Transport(msg.clone())
        }
        ConnectorError::BadRequestError { msg, code }
        | ConnectorError::NotFoundError { msg, code }
        | ConnectorError::ConnectorClientError { msg, code } => {
            // A codeless `ConnectorClientError` from the SDK's `http_request` is a transport or
            // response-decode failure, not a venue decision (genuine Binance rejections carry a
            // numeric code). These prefixes are the SDK's codeless transport/decode sites: the
            // request never completed, or a 2xx body could not be read/decompressed/decoded.
            // Route them to Transport — the request's venue status is unknown — rather than
            // misreporting a definitive rejection. Match by prefix (not a blanket `code.is_none()`)
            // so an unusual HTTP error *status* with no Binance code still maps to a rejection.
            // (Body-deserialization failures surface via `.data()`, a separate path handled at
            // that call site.)
            //
            // These prefixes mirror the codeless error sites in the binance-sdk `http_request`
            // helper (binance-sdk `src/common`). They are not a stable public contract — re-verify
            // this list (and the test below that pins it) whenever the binance-sdk pin is bumped.
            const TRANSPORT_PREFIXES: [&str; 4] = [
                "HTTP request failed",
                "Failed to get response bytes",
                "Failed to decompress gzip response",
                "Failed to convert response to UTF-8",
            ];
            if code.is_none() && TRANSPORT_PREFIXES.iter().any(|p| msg.starts_with(p)) {
                RequestFailure::Transport(msg.clone())
            } else {
                RequestFailure::Rejected(with_code(msg, *code))
            }
        }
    }
}

/// Classify an `anyhow::Error` from a REST order/cancel call into an [`OrderError`].
///
/// REST errors differ from the WS-API path: binance-sdk surfaces them as
/// [`ConnectorError`], whose `Display` **omits** the Binance numeric code (it lives in a
/// separate `code` field). So the WS classifier [`classify_ws_order_error`] (which downcasts
/// to `WebsocketError`) does not apply here — we downcast to `ConnectorError` instead, classify
/// it with [`classify_connector_error`], and map a venue rejection with
/// [`parse_binance_order_rejection`].
///
/// - [`CooldownActive`] (held back unsent) → [`ApiError::RateLimit`], with the cooldown's time
///   left as the wait.
/// - 401 → [`ApiError::Unauthenticated`]; 429/418 → [`ApiError::RateLimit`], with Binance's
///   `Retry-After` as the wait; 403 → `Unauthenticated` when it is an auth failure, else
///   `RateLimit` (Binance's WAF limit), with no wait.
/// - 400/404 / other client errors that carry a Binance code → mapped by code/text; a venue
///   failure code → [`OrderError::Connectivity`].
/// - Network/server failures (and the SDK's codeless transport/decode wrappers — failed HTTP
///   request, response-byte read, gzip, or UTF-8 decode) → [`OrderError::Connectivity`]: the
///   order may or may not have reached the matching engine.
pub(crate) fn classify_rest_order_error(
    e: &anyhow::Error,
    instrument: &InstrumentNameExchange,
) -> OrderError<AssetNameExchange, InstrumentNameExchange> {
    if let Some(cooldown) = e.downcast_ref::<CooldownActive>() {
        return OrderError::Rejected(ApiError::rate_limited(Some(cooldown.remaining)));
    }
    let Some(ce) = e.downcast_ref::<ConnectorError>() else {
        // Not an SDK ConnectorError — treat as opaque transport failure.
        return OrderError::Connectivity(ConnectivityError::Socket(format!("{e:#}")));
    };

    order_error_from(classify_connector_error(ce), instrument)
}

/// Classify an `anyhow::Error` from a WS-API order/cancel request.
///
/// binance-sdk returns both transport failures and venue responses with status >= 400 through
/// the same `Err`. Returns `None` unless it is a [`WebsocketError::ResponseError`]: the venue
/// answered, so the session is healthy and the caller keeps it. Any other error is a transport
/// failure for the caller to handle. `downcast_ref` searches the whole error chain, so context
/// layers do not hide the `ResponseError`. Re-verify on an SDK upgrade: if the SDK changes its
/// error wrapping, `downcast_ref` returns `None` and every rejection would be treated as a
/// transport failure.
pub(crate) fn classify_ws_order_error(
    e: &anyhow::Error,
    instrument: &InstrumentNameExchange,
) -> Option<UnindexedOrderError> {
    ws_response_failure(e).map(|failure| order_error_from(failure, instrument))
}

/// Classify `e` when it is a WS-API [`WebsocketError::ResponseError`], the venue answering a
/// request with a status of 400 or more; `None` for any other error. `downcast_ref` also finds it
/// beneath context layers.
fn ws_response_failure(e: &anyhow::Error) -> Option<RequestFailure> {
    match e.downcast_ref::<WebsocketError>()? {
        // The typed error's own `Display`, not `e`'s: a context layer would hide the code.
        response @ WebsocketError::ResponseError {
            code, retry_after, ..
        } => Some(classify_ws_response_error(
            *code,
            // A ban that has already lifted, by the local clock, names no wait: the default
            // cooldown applies, as it does to a refusal that named none.
            retry_after
                .map(|until_ms| wait_until_epoch_ms(until_ms, unix_ms()))
                .filter(|wait| !wait.is_zero()),
            response.to_string(),
        )),
        _ => None,
    }
}

/// Map a [`RequestFailure`] of an order or cancel request to an [`OrderError`].
fn order_error_from(
    failure: RequestFailure,
    instrument: &InstrumentNameExchange,
) -> UnindexedOrderError {
    match failure {
        RequestFailure::RateLimited(retry_after) => {
            OrderError::Rejected(ApiError::rate_limited(retry_after))
        }
        RequestFailure::Unauthenticated(msg) => {
            OrderError::Rejected(ApiError::Unauthenticated(msg))
        }
        RequestFailure::Transport(msg) => OrderError::Connectivity(ConnectivityError::Socket(msg)),
        RequestFailure::Rejected(msg) => parse_binance_order_rejection(msg, instrument),
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] // Test code: panics on bad input are acceptable
mod tests {
    use super::*;
    use crate::client::reconnect_budget::BackoffLimits;
    use futures::FutureExt as _;

    /// The module docs and both clients' rustdoc state the heartbeat window and the longest
    /// cooldown waited through as numbers. Change these only together with those docs. The
    /// reconnect budget's own numbers are pinned in `client::reconnect_budget`.
    #[test]
    fn the_heartbeat_and_cooldown_are_as_documented() {
        assert_eq!(HEARTBEAT_TIMEOUT_SECS, 30);
        assert_eq!(MAX_COOLDOWN_SLEEP, Duration::from_secs(10));
    }

    /// A connection that counts its closes, and whose close never finishes while `stall` is set.
    #[derive(Clone, Default)]
    struct CountingWs {
        closes: Arc<AtomicU32>,
        stall: Arc<std::sync::atomic::AtomicBool>,
    }

    impl BinanceWsConnection for CountingWs {
        async fn close(&self) -> anyhow::Result<()> {
            self.closes.fetch_add(1, Ordering::SeqCst);
            if self.stall.load(Ordering::SeqCst) {
                std::future::pending::<()>().await;
            }
            Ok(())
        }
    }

    impl CountingWs {
        /// Closes so far, once every task the guard spawned has had a chance to run.
        async fn closes(&self) -> u32 {
            for _ in 0..10 {
                tokio::task::yield_now().await;
            }
            self.closes.load(Ordering::SeqCst)
        }
    }

    #[tokio::test]
    async fn a_dropped_guard_closes_its_connection() {
        let ws = CountingWs::default();
        drop(DisconnectOnDrop::new(ws.clone(), "test"));
        assert_eq!(ws.closes().await, 1);
    }

    #[tokio::test]
    async fn a_disconnected_guard_does_not_close_again_on_drop() {
        let ws = CountingWs::default();
        let result = DisconnectOnDrop::new(ws.clone(), "test").disconnect().await;
        assert!(result.is_ok());
        assert_eq!(ws.closes().await, 1);
    }

    /// A disconnect abandoned part-way, by a timeout or an abort, leaves the connection to the
    /// guard's drop, which closes it again.
    #[tokio::test]
    async fn an_abandoned_disconnect_closes_again_on_drop() {
        let ws = CountingWs::default();
        ws.stall.store(true, Ordering::SeqCst);
        let disconnect = DisconnectOnDrop::new(ws.clone(), "test").disconnect();
        assert!(
            tokio::time::timeout(Duration::from_millis(10), disconnect)
                .await
                .is_err()
        );
        assert_eq!(ws.closes().await, 2);
    }

    /// While its event emitter lives, binance-sdk ends a subscription's callback task only on
    /// unsubscribe, so the guard unsubscribes when dropped.
    #[tokio::test]
    async fn a_dropped_subscription_guard_stops_its_callback() {
        let emitter = binance_sdk::common::websocket::WebsocketEventEmitter::new();
        let callback_state = Arc::new(());
        let held = Arc::clone(&callback_state);
        let subscription = emitter.subscribe(move |_| {
            let _ = &held;
        });

        drop(UnsubscribeOnDrop::new(subscription));
        for _ in 0..10 {
            tokio::task::yield_now().await;
        }

        assert_eq!(
            Arc::strong_count(&callback_state),
            1,
            "the callback was dropped"
        );
    }

    /// Live tasks on the current runtime.
    fn alive() -> usize {
        tokio::runtime::Handle::current()
            .metrics()
            .num_alive_tasks()
    }

    /// A WS-API config for a local server that accepts every connection and answers nothing.
    async fn local_ws_api_config() -> binance_sdk::common::config::ConfigurationWebsocketApi {
        let (url, _) = super::test_ws_api::serve(
            |_| serde_json::Value::Null,
            |_| super::test_ws_api::Lifetime::Open,
        )
        .await;
        binance_sdk::common::config::ConfigurationWebsocketApi::builder()
            .api_key("key")
            .api_secret("secret")
            .ws_url(url)
            .build()
            .unwrap()
    }

    /// Asserts the live tasks fall back to `before`. The server ends its side of each connection
    /// once the close reaches it, so this waits up to 5 s.
    async fn assert_tasks_return_to(before: usize) {
        let settled = tokio::time::timeout(Duration::from_secs(5), async {
            while alive() > before {
                tokio::time::sleep(Duration::from_millis(20)).await;
            }
        })
        .await;
        assert!(
            settled.is_ok(),
            "tasks left behind: {} of {before}",
            alive()
        );
        assert_eq!(alive(), before);
    }

    /// binance-sdk frees a connection once it is disconnected and dropped. Before 76.0.0 its
    /// reconnect and renewal loops kept it alive, leaking two tasks for each order session
    /// reopened and each account stream reconnected. With no task left, none is left that could
    /// reopen the connection either, so this also pins that a disconnected connection stays closed.
    #[tokio::test]
    async fn a_disconnected_connection_leaves_no_task_behind() {
        let config = local_ws_api_config().await;
        let before = alive();

        // The spot order session, and the common-layer connection the margin account stream uses.
        let session = binance_sdk::spot::SpotWsApi::from_config(config.clone())
            .connect()
            .await
            .unwrap();
        DisconnectOnDrop::new(session, "spot")
            .disconnect()
            .await
            .unwrap();
        let stream = binance_sdk::common::websocket::WebsocketApi::new(config, vec![]);
        Arc::clone(&stream).connect().await.unwrap();
        DisconnectOnDrop::new(stream, "margin")
            .disconnect()
            .await
            .unwrap();

        assert_tasks_return_to(before).await;
    }

    /// A connection dropped without being disconnected is freed too: a guard dropped unclosed,
    /// as when a connection manager is aborted with its stream, and a bare SDK connection whose
    /// event callback was never unsubscribed.
    #[tokio::test]
    async fn a_dropped_connection_leaves_no_task_behind() {
        let config = local_ws_api_config().await;
        let before = alive();

        let session = binance_sdk::spot::SpotWsApi::from_config(config.clone())
            .connect()
            .await
            .unwrap();
        drop(DisconnectOnDrop::new(session, "spot"));
        let stream = binance_sdk::common::websocket::WebsocketApi::new(config, vec![]);
        Arc::clone(&stream).connect().await.unwrap();
        let callback_state = Arc::new(());
        let held = Arc::clone(&callback_state);
        let _subscription = stream.common.events.subscribe(move |_| {
            let _ = &held;
        });
        drop(stream);

        assert_tasks_return_to(before).await;
        assert_eq!(
            Arc::strong_count(&callback_state),
            1,
            "the callback was dropped"
        );
    }

    /// Outside a runtime the guard cannot spawn the close; it logs instead of panicking.
    #[test]
    fn a_guard_dropped_outside_a_runtime_does_not_panic() {
        let ws = CountingWs::default();
        drop(DisconnectOnDrop::new(ws.clone(), "test"));
        assert_eq!(ws.closes.load(Ordering::SeqCst), 0);
    }

    /// The loud path runs for the first event of each type and for every 1000th event overall, so
    /// a frequent unhandled type cannot hide a later one; repeats of a type stay quiet.
    #[test]
    fn unhandled_events_warn_once_per_type_and_every_thousandth() {
        let seen = UnhandledEvents::new();
        assert_eq!(seen.record("frequent"), (1, true));
        for count in 2..=10 {
            assert_eq!(seen.record("frequent"), (count, false));
        }
        assert_eq!(
            seen.record("renamed"),
            (11, true),
            "a new type warns despite the first"
        );
        assert_eq!(seen.record("renamed"), (12, false));
        for count in 13..1000 {
            assert_eq!(seen.record("frequent"), (count, false));
        }
        assert_eq!(seen.record("frequent"), (1000, true));

        // Past the cap a new type is no longer tracked, so it stays quiet between thousandths.
        let capped = UnhandledEvents::new();
        for i in 0..UnhandledEvents::MAX_TYPES {
            assert!(capped.record(&format!("type{i}")).1);
        }
        assert!(!capped.record("one_too_many").1);
    }

    fn report(json: &str) -> ExecutionReport {
        match serde_json::from_str(json) {
            Ok(report) => report,
            Err(error) => panic!("{json} should parse: {error}"),
        }
    }

    /// `N` reads as no commission asset whether absent or `null`, and a value reads as itself.
    #[test]
    fn a_report_reads_an_absent_null_or_present_commission_asset() {
        assert_eq!(report(r#"{"e":"executionReport"}"#).commission_asset, None);
        assert_eq!(report(r#"{"N":null}"#).commission_asset, None);
        assert_eq!(
            report(r#"{"N":"BNB"}"#).commission_asset.as_deref(),
            Some("BNB")
        );
    }

    /// `d` decodes whether Binance sends it as a number, as its docs show, or as a string, as the
    /// margin SDK model declared until 74.1.1; anything else is kept as unreadable rather than
    /// failing the report.
    #[test]
    fn a_report_reads_the_trailing_delta_as_a_number_or_a_string() {
        assert_eq!(report(r#"{}"#).trailing_delta, None);
        assert_eq!(report(r#"{"d":null}"#).trailing_delta, None);
        assert_eq!(report(r#"{"d":25}"#).trailing_delta, Some(WireInt::Int(25)));
        assert_eq!(
            report(r#"{"d":"25"}"#).trailing_delta,
            Some(WireInt::Int(25))
        );
        assert_eq!(
            report(r#"{"d":"x"}"#).trailing_delta,
            Some(WireInt::Unreadable("x".to_owned()))
        );
        assert_eq!(
            report(r#"{"d":true}"#).trailing_delta,
            Some(WireInt::Unreadable("true".to_owned()))
        );
        assert_eq!(
            report(r#"{"d":[1],"x":"NEW"}"#).trailing_delta,
            Some(WireInt::Unreadable("an array".to_owned()))
        );
        assert_eq!(
            report(r#"{"d":{"a":1},"x":"NEW"}"#)
                .execution_type
                .as_deref(),
            Some("NEW")
        );
        assert_eq!(
            report(r#"{"d":2.5}"#).trailing_delta,
            Some(WireInt::Unreadable("2.5".to_owned()))
        );
    }

    /// Keys the converter does not read cannot fail a report, whatever their type: the keys whose
    /// type binance-sdk's margin model has changed (`D`, `j`, `J`, `u`, `U`, `a`, `v`) decode as
    /// numbers, as strings and as `null`, and the keys it reads come through each time.
    #[test]
    fn keys_the_converter_does_not_read_cannot_fail_a_report() {
        for (unread, delta) in [
            (
                r#""D":1668680518494,"j":1,"J":1000000,"u":2,"U":3,"a":5,"v":3"#,
                "4",
            ),
            (
                r#""D":"1668680518494","j":"1","J":"1000000","u":"2","U":"3","a":"5","v":"3""#,
                r#""4""#,
            ),
            (
                r#""D":null,"j":null,"J":null,"u":null,"U":null,"a":null,"v":null"#,
                "4",
            ),
        ] {
            let json = format!(
                r#"{{"e":"executionReport","x":"NEW","i":7,"o":"STOP_LOSS","P":"0","d":{delta},{unread},"W":1499405658658,"V":"NONE"}}"#
            );
            let parsed = report(&json);
            assert_eq!(parsed.execution_type.as_deref(), Some("NEW"), "{json}");
            assert_eq!(parsed.order_id, Some(7), "{json}");
            assert_eq!(parsed.trailing_delta, Some(WireInt::Int(4)), "{json}");
        }
    }

    #[test]
    fn a_reported_trailing_delta_reads_absent_when_missing_or_zero() {
        assert_eq!(TrailingDelta::from_reported(None), TrailingDelta::Absent);
        assert_eq!(TrailingDelta::from_reported(Some(0)), TrailingDelta::Absent);
        assert_eq!(
            TrailingDelta::from_reported(Some(25)),
            TrailingDelta::BasisPoints(25)
        );
        assert_eq!(
            TrailingDelta::from_reported(Some(-1)),
            TrailingDelta::Unknown
        );
    }

    fn decode(
        order_type: &str,
        stop_price: Option<&str>,
        trailing_delta: TrailingDelta,
    ) -> Option<OrderKind> {
        decode_order_kind(
            ExchangeId::BinanceSpot,
            Some(1),
            order_type,
            stop_price,
            trailing_delta,
        )
    }

    #[test]
    fn plain_types_decode_whatever_the_trailing_delta() {
        for delta in [
            TrailingDelta::Absent,
            TrailingDelta::BasisPoints(25),
            TrailingDelta::Unknown,
        ] {
            assert_eq!(decode("MARKET", None, delta), Some(OrderKind::Market));
            assert_eq!(decode("LIMIT", None, delta), Some(OrderKind::Limit));
            assert_eq!(decode("LIMIT_MAKER", None, delta), Some(OrderKind::Limit));
        }
        assert_eq!(decode("UNKNOWN_TYPE", None, TrailingDelta::Absent), None);
    }

    #[test]
    fn a_conditional_order_without_a_trailing_delta_decodes_with_its_trigger() {
        let trigger_price = Decimal::from(100);
        let cases = [
            ("STOP_LOSS", OrderKind::Stop { trigger_price }),
            ("STOP_LOSS_LIMIT", OrderKind::StopLimit { trigger_price }),
            ("TAKE_PROFIT", OrderKind::TakeProfit { trigger_price }),
            (
                "TAKE_PROFIT_LIMIT",
                OrderKind::TakeProfitLimit { trigger_price },
            ),
        ];
        for (order_type, kind) in cases {
            assert_eq!(
                decode(order_type, Some("100.00"), TrailingDelta::Absent),
                Some(kind),
                "{order_type}"
            );
            // Without a positive trigger the order cannot be described.
            for stop_price in [None, Some("0.00000000"), Some("-1"), Some("x")] {
                assert_eq!(
                    decode(order_type, stop_price, TrailingDelta::Absent),
                    None,
                    "{order_type} {stop_price:?}"
                );
            }
        }
    }

    /// A trailing `STOP_LOSS` decodes in basis points, and a trailing `STOP_LOSS_LIMIT` as a
    /// fixed-limit trailing stop, each with its positive `stopPrice` as the activation price.
    #[test]
    fn a_trailing_stop_loss_decodes_with_its_activation_price() {
        use rust_decimal_macros::dec;
        for (stop_price, activation_price) in [
            (None, None),
            (Some("0.00000000"), None),
            (Some("105.5"), Some(dec!(105.5))),
        ] {
            assert_eq!(
                decode("STOP_LOSS", stop_price, TrailingDelta::BasisPoints(25)),
                Some(OrderKind::TrailingStop {
                    offset: Decimal::from(25),
                    offset_type: TrailingOffsetType::BasisPoints,
                    activation_price,
                }),
                "{stop_price:?}"
            );
            assert_eq!(
                decode(
                    "STOP_LOSS_LIMIT",
                    stop_price,
                    TrailingDelta::BasisPoints(25)
                ),
                Some(OrderKind::TrailingStopFixedLimit {
                    offset: Decimal::from(25),
                    offset_type: TrailingOffsetType::BasisPoints,
                    activation_price,
                }),
                "{stop_price:?}"
            );
        }
    }

    /// A trailing take-profit has no `OrderKind`, a conditional order whose trailing delta is
    /// unknown could be either fixed or trailing, and a trailing order whose stopPrice does not
    /// parse has an unknown activation price: all are dropped rather than misdescribed.
    #[test]
    fn conditional_orders_that_cannot_be_described_are_dropped() {
        for order_type in ["STOP_LOSS", "STOP_LOSS_LIMIT"] {
            assert_eq!(
                decode(order_type, Some("x"), TrailingDelta::BasisPoints(25)),
                None,
                "{order_type}"
            );
        }
        for order_type in ["TAKE_PROFIT", "TAKE_PROFIT_LIMIT"] {
            assert_eq!(
                decode(order_type, Some("100"), TrailingDelta::BasisPoints(25)),
                None,
                "{order_type}"
            );
        }
        for order_type in [
            "STOP_LOSS",
            "STOP_LOSS_LIMIT",
            "TAKE_PROFIT",
            "TAKE_PROFIT_LIMIT",
        ] {
            assert_eq!(
                decode(order_type, Some("100"), TrailingDelta::Unknown),
                None,
                "{order_type}"
            );
        }
    }

    #[test]
    fn a_trailing_offset_converts_to_whole_basis_points() {
        use rust_decimal_macros::dec;
        let ok = |offset, offset_type| trailing_delta_basis_points(offset, offset_type);
        assert_eq!(ok(dec!(25), TrailingOffsetType::BasisPoints), Ok(25));
        assert_eq!(ok(dec!(25.0), TrailingOffsetType::BasisPoints), Ok(25));
        assert_eq!(ok(dec!(1.5), TrailingOffsetType::Percentage), Ok(150));
        assert_eq!(ok(dec!(0.15), TrailingOffsetType::Percentage), Ok(15));
        assert_eq!(
            ok(dec!(1), TrailingOffsetType::Absolute),
            Err(TrailingDeltaError::AbsoluteOffset)
        );
    }

    /// An offset that is not a positive whole number of basis points is refused, never rounded.
    #[test]
    fn a_trailing_offset_that_is_not_whole_positive_basis_points_is_refused() {
        use rust_decimal_macros::dec;
        let refused = |offset, offset_type, limit| {
            assert_eq!(
                trailing_delta_basis_points(offset, offset_type),
                Err(TrailingDeltaError::Invalid(PrecisionViolation::new(
                    OrderField::TrailingOffset,
                    offset,
                    limit
                ))),
                "{offset} {offset_type:?}"
            );
        };
        refused(
            dec!(0.155),
            TrailingOffsetType::Percentage,
            PrecisionLimit::DecimalPlaces { max: 2 },
        );
        refused(
            dec!(2.5),
            TrailingOffsetType::BasisPoints,
            PrecisionLimit::DecimalPlaces { max: 0 },
        );
        for offset_type in [
            TrailingOffsetType::BasisPoints,
            TrailingOffsetType::Percentage,
        ] {
            refused(dec!(0), offset_type, PrecisionLimit::NotPositive);
            refused(dec!(-10), offset_type, PrecisionLimit::NotPositive);
        }
        refused(
            Decimal::from(i64::from(i32::MAX) + 1),
            TrailingOffsetType::BasisPoints,
            PrecisionLimit::NotRepresentable,
        );
        refused(
            Decimal::MAX,
            TrailingOffsetType::Percentage,
            PrecisionLimit::NotRepresentable,
        );
    }

    #[test]
    fn a_trailing_offset_refusal_is_an_order_error() {
        use rust_decimal_macros::dec;
        let violation = PrecisionViolation::new(
            OrderField::TrailingOffset,
            dec!(2.5),
            PrecisionLimit::NotPositive,
        );
        assert_eq!(
            TrailingDeltaError::Invalid(violation).into_order_error::<(), ()>(),
            OrderError::InvalidPrecision(violation)
        );
        assert!(matches!(
            TrailingDeltaError::AbsoluteOffset.into_order_error::<(), ()>(),
            OrderError::UnsupportedOrderType(_)
        ));
    }

    /// A REST order row of `order_type`, with `extra` keys merged in.
    fn order_row(order_type: &str, status: &str, extra: serde_json::Value) -> serde_json::Value {
        let mut row = serde_json::json!({
            "symbol": "BTCUSDT", "orderId": 7, "clientOrderId": "cid-7", "price": "0.00000000",
            "origQty": "2", "executedQty": "0", "cummulativeQuoteQty": "0", "status": status,
            "timeInForce": "GTC", "type": order_type, "side": "SELL",
            "stopPrice": "0.00000000", "time": 1_700_000_000_000_i64,
            "updateTime": 1_700_000_060_000_i64,
        });
        if let (Some(row), serde_json::Value::Object(extra)) = (row.as_object_mut(), extra) {
            row.extend(extra);
        }
        row
    }

    fn spot_row(row: serde_json::Value) -> binance_sdk::spot::rest_api::GetOpenOrdersResponseInner {
        match serde_json::from_value(row) {
            Ok(row) => row,
            Err(error) => panic!("spot row should parse: {error}"),
        }
    }

    fn margin_row(
        row: serde_json::Value,
    ) -> binance_sdk::margin_trading::rest_api::QueryMarginAccountsOpenOrdersResponseInner {
        match serde_json::from_value(row) {
            Ok(row) => row,
            Err(error) => panic!("margin row should parse: {error}"),
        }
    }

    /// A trailing STOP_LOSS_LIMIT row lists as a fixed-limit trailing stop with its limit price;
    /// a zero stopPrice is no activation price.
    #[test]
    fn a_trailing_stop_loss_limit_row_lists_with_its_limit_price() {
        let btc = InstrumentNameExchange::new("BTCUSDT");
        for (stop_price, activation_price) in
            [("0.00000000", None), ("105", Some(Decimal::from(105)))]
        {
            let rows = [spot_row(order_row(
                "STOP_LOSS_LIMIT",
                "NEW",
                serde_json::json!({
                    "trailingDelta": 25, "trailingTime": -1, "price": "95", "stopPrice": stop_price,
                }),
            ))];
            let listing = convert_open_order_listing(&rows, ExchangeId::BinanceSpot, &btc);
            let [order] = listing.orders.as_slice() else {
                panic!("one order: {:?}", listing.orders);
            };
            assert_eq!(
                order.kind,
                OrderKind::TrailingStopFixedLimit {
                    offset: Decimal::from(25),
                    offset_type: TrailingOffsetType::BasisPoints,
                    activation_price,
                }
            );
            assert_eq!(order.price, Some(Decimal::from(95)));
        }
    }

    /// A trailing order's fields: its delta, a fixed limit's price, an activation price; and the
    /// requests refused unsent.
    #[test]
    fn trailing_params_carry_the_delta_limit_and_activation_and_refuse_bad_prices() {
        use rust_decimal_macros::dec;
        let fixed_limit = |activation_price| OrderKind::TrailingStopFixedLimit {
            offset: dec!(1.5),
            offset_type: TrailingOffsetType::Percentage,
            activation_price,
        };
        assert_eq!(trailing_params(&OrderKind::Limit, Some(dec!(1))), Ok(None));
        assert_eq!(
            trailing_params(&fixed_limit(Some(dec!(105))), Some(dec!(99))),
            Ok(Some(TrailingParams {
                trailing_delta: 150,
                price: Some(dec!(99)),
                stop_price: Some(dec!(105)),
            }))
        );
        assert_eq!(
            trailing_params(
                &OrderKind::TrailingStop {
                    offset: dec!(25),
                    offset_type: TrailingOffsetType::BasisPoints,
                    activation_price: None,
                },
                Some(dec!(99))
            ),
            Ok(Some(TrailingParams {
                trailing_delta: 25,
                price: None,
                stop_price: None,
            })),
            "a trailing stop sends no price"
        );
        for activation in [dec!(0), dec!(-1)] {
            assert_eq!(
                trailing_params(&fixed_limit(Some(activation)), Some(dec!(99))),
                Err(TrailingOrderError::NonPositiveActivation(activation))
            );
        }
        for limit in [None, Some(dec!(0))] {
            assert_eq!(
                trailing_params(&fixed_limit(None), limit),
                Err(TrailingOrderError::LimitPrice(limit))
            );
        }
        // A bad offset is reported before a bad price.
        assert!(matches!(
            trailing_params(
                &OrderKind::TrailingStopFixedLimit {
                    offset: dec!(1.555),
                    offset_type: TrailingOffsetType::Percentage,
                    activation_price: Some(dec!(0)),
                },
                None
            ),
            Err(TrailingOrderError::Delta(TrailingDeltaError::Invalid(_)))
        ));
        assert!(matches!(
            TrailingOrderError::LimitPrice(None).into_order_error::<(), ()>(),
            OrderError::Rejected(ApiError::OrderRejected(_))
        ));
        assert!(matches!(
            TrailingOrderError::NonPositiveActivation(dec!(0)).into_order_error::<(), ()>(),
            OrderError::Rejected(ApiError::OrderRejected(_))
        ));
        assert!(matches!(
            TrailingOrderError::Delta(TrailingDeltaError::AbsoluteOffset)
                .into_order_error::<(), ()>(),
            OrderError::UnsupportedOrderType(_)
        ));
    }

    /// Spot rows report `trailingDelta`, so a trailing stop lists as one, and conditional orders
    /// list with their triggers, with no limit price where the kind takes none.
    #[test]
    fn spot_rows_list_conditional_and_trailing_orders() {
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let rows = [
            spot_row(order_row(
                "STOP_LOSS",
                "NEW",
                serde_json::json!({"trailingDelta": 25, "trailingTime": -1}),
            )),
            spot_row(order_row(
                "STOP_LOSS_LIMIT",
                "NEW",
                serde_json::json!({"clientOrderId": "cid-8", "price": "95", "stopPrice": "96"}),
            )),
            spot_row(order_row(
                "STOP_LOSS",
                "NEW",
                serde_json::json!({"clientOrderId": "cid-9", "stopPrice": "96"}),
            )),
        ];
        let listing = convert_open_order_listing(&rows, ExchangeId::BinanceSpot, &btc);
        assert!(listing.complete);
        let described: Vec<_> = listing
            .orders
            .iter()
            .map(|order| (order.key.cid.0.as_str(), order.kind, order.price))
            .collect();
        assert_eq!(
            described,
            [
                (
                    "cid-7",
                    OrderKind::TrailingStop {
                        offset: Decimal::from(25),
                        offset_type: TrailingOffsetType::BasisPoints,
                        activation_price: None,
                    },
                    None,
                ),
                (
                    "cid-8",
                    OrderKind::StopLimit {
                        trigger_price: Decimal::from(96)
                    },
                    Some(Decimal::from(95)),
                ),
                (
                    "cid-9",
                    OrderKind::Stop {
                        trigger_price: Decimal::from(96)
                    },
                    None,
                ),
            ]
        );
    }

    /// Margin rows do not report `trailingDelta`, so a live conditional row could be a trailing
    /// order: it is left out and the listing reads incomplete, while plain rows list as before.
    #[test]
    fn margin_listings_leave_out_conditional_rows_and_read_incomplete() {
        let btc = InstrumentNameExchange::new("BTCUSDT");
        for order_type in [
            "STOP_LOSS",
            "STOP_LOSS_LIMIT",
            "TAKE_PROFIT",
            "TAKE_PROFIT_LIMIT",
        ] {
            let rows = [
                margin_row(order_row(
                    "LIMIT",
                    "NEW",
                    serde_json::json!({"clientOrderId": "cid-1", "price": "95"}),
                )),
                margin_row(order_row(
                    order_type,
                    "NEW",
                    serde_json::json!({"price": "95", "stopPrice": "96"}),
                )),
            ];
            let listing = convert_open_order_listing(&rows, ExchangeId::BinanceMargin, &btc);
            assert!(!listing.complete, "{order_type}");
            assert_eq!(listing.orders.len(), 1, "{order_type}");
            assert_eq!(listing.orders[0].kind, OrderKind::Limit);
        }
    }

    /// A margin conditional order that has ended is still reported as ended, so it is not held as
    /// live: with the kind the client holds it with, or failing that with its fixed trigger,
    /// without which it cannot be described. The held kind stands in only for what the row cannot
    /// tell.
    #[test]
    fn an_ended_margin_conditional_order_reports_its_ending() {
        let key = UnindexedOrderKey::new(
            ExchangeId::BinanceMargin,
            InstrumentNameExchange::new("BTCUSDT"),
            StrategyId::unknown(),
            ClientOrderId::new("cid-7"),
        );
        let row = margin_row(order_row(
            "STOP_LOSS",
            "CANCELED",
            serde_json::json!({"stopPrice": "96"}),
        ));
        let Some(order) = convert_ended_order(&row, ExchangeId::BinanceMargin, &key, None) else {
            panic!("an ended stop should be reported");
        };
        assert_eq!(
            order.kind,
            OrderKind::Stop {
                trigger_price: Decimal::from(96)
            }
        );
        assert!(matches!(order.state, InactiveOrderState::Cancelled(_)));

        let untriggered = margin_row(order_row("STOP_LOSS", "CANCELED", serde_json::json!({})));
        assert!(convert_ended_order(&untriggered, ExchangeId::BinanceMargin, &key, None).is_none());

        // With the kind the client holds the order with, that kind is reported, trigger or not.
        let trailing_stop = OrderKind::TrailingStop {
            offset: Decimal::from(25),
            offset_type: TrailingOffsetType::BasisPoints,
            activation_price: None,
        };
        for row in [&row, &untriggered] {
            let Some(order) =
                convert_ended_order(row, ExchangeId::BinanceMargin, &key, Some(trailing_stop))
            else {
                panic!("an ended order held with its kind should be reported");
            };
            assert_eq!(order.kind, trailing_stop);
            assert_eq!(order.price, None);
        }

        // A row that can describe the order is read as it describes it, whatever kind the client
        // holds: a margin `LIMIT` row, and a spot stop row, which reports `trailingDelta`.
        let limit = margin_row(order_row(
            "LIMIT",
            "CANCELED",
            serde_json::json!({"price": "95"}),
        ));
        let spot_stop = spot_row(order_row(
            "STOP_LOSS",
            "CANCELED",
            serde_json::json!({"stopPrice": "96"}),
        ));
        let read = [
            convert_ended_order(&limit, ExchangeId::BinanceMargin, &key, Some(trailing_stop)),
            convert_ended_order(
                &spot_stop,
                ExchangeId::BinanceSpot,
                &key,
                Some(trailing_stop),
            ),
        ]
        .map(|order| order.map(|order| (order.kind, order.price)));
        assert_eq!(
            read,
            [
                Some((OrderKind::Limit, Some(Decimal::from(95)))),
                Some((
                    OrderKind::Stop {
                        trigger_price: Decimal::from(96)
                    },
                    None
                )),
            ]
        );
    }

    /// A gap is due once opened, leaves when recovered, and when its read fails is retried with a
    /// doubling delay until it has failed every retry, then given up.
    #[tokio::test]
    async fn a_fill_gap_is_retried_with_backoff_then_given_up() {
        tokio::time::pause();
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let eth = InstrumentNameExchange::new("ETHUSDT");
        let disconnect = Utc.timestamp_millis_opt(1_000_000).unwrap();
        let reconnect = Utc.timestamp_millis_opt(2_000_000).unwrap();
        let mut unrecovered = UnrecoveredFills::default();
        unrecovered.open(&[btc.clone(), eth.clone()], disconnect, reconnect);

        let due = unrecovered.due(tokio::time::Instant::now());
        assert_eq!(due.len(), 2);
        let gap_of = |instrument: &InstrumentNameExchange| {
            let Some((_, gap)) = due.iter().find(|(inst, _)| inst == instrument) else {
                panic!("{instrument} has a gap");
            };
            *gap
        };
        let gap = gap_of(&btc);
        assert_eq!(gap.start_ms, 1_000_000);
        assert_eq!(gap.end_ms, 2_000_000 + GAP_END_SLACK_SECS * 1_000);

        unrecovered.recovered(&eth, &gap_of(&eth));
        for retry in 0..MAX_GAP_RETRIES {
            let delay = Duration::from_secs(GAP_RETRY_BASE_SECS << retry);
            let now = tokio::time::Instant::now();
            assert_eq!(
                unrecovered.failed(&btc, &gap, now),
                Some(GapFailure::Retry(delay))
            );
            assert!(unrecovered.due(now).is_empty(), "not due before its delay");
            assert_eq!(unrecovered.next_due(), Some(now + delay));
            tokio::time::advance(delay).await;
            assert_eq!(unrecovered.due(tokio::time::Instant::now()).len(), 1);
        }
        assert_eq!(
            unrecovered.failed(&btc, &gap, tokio::time::Instant::now()),
            Some(GapFailure::GivenUp)
        );
        assert!(unrecovered.is_empty());
        assert_eq!(unrecovered.next_due(), None);
        assert_eq!(
            unrecovered.failed(&btc, &gap, tokio::time::Instant::now()),
            None
        );
    }

    /// A gap is reported on the stream only when it is given up, after its first read and every
    /// retry failed, with its instrument, its span, how many reads failed and the last failure.
    #[tokio::test]
    async fn a_given_up_fill_gap_is_reported_on_the_stream() {
        tokio::time::pause();
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let disconnect = Utc.timestamp_millis_opt(1_000_000).unwrap();
        let reconnect = Utc.timestamp_millis_opt(2_000_000).unwrap();
        let mut unrecovered = UnrecoveredFills::default();
        unrecovered.open(std::slice::from_ref(&btc), disconnect, reconnect);
        let Some((_, gap)) = unrecovered.due(tokio::time::Instant::now()).pop() else {
            panic!("the gap is due");
        };
        let (tx, mut rx) = mpsc::unbounded_channel();

        for _ in 0..MAX_GAP_RETRIES {
            gap_failed(
                &mut unrecovered,
                &tx,
                ExchangeId::BinanceSpot,
                &btc,
                &gap,
                FillRecoveryFailure::Request("rejected".to_string()),
            );
            assert!(
                rx.try_recv().is_err(),
                "a gap left to retry is not reported"
            );
            unrecovered.make_due();
        }
        gap_failed(
            &mut unrecovered,
            &tx,
            ExchangeId::BinanceSpot,
            &btc,
            &gap,
            FillRecoveryFailure::TimedOut { timeout_secs: 30 },
        );

        let Ok(event) = rx.try_recv() else {
            panic!("the given-up gap is reported");
        };
        assert_eq!(event.exchange, ExchangeId::BinanceSpot);
        assert_eq!(
            event.kind,
            AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
                FillRecoveryScope::Instruments(vec![btc]),
                disconnect,
                reconnect + chrono::Duration::seconds(GAP_END_SLACK_SECS),
                MAX_GAP_RETRIES + 1,
                FillRecoveryFailure::TimedOut { timeout_secs: 30 },
            ))
        );
        assert!(rx.try_recv().is_err(), "reported once");
        assert!(unrecovered.is_empty(), "and no longer kept");
    }

    /// A new gap that overlaps a kept one starts after it, so the kept one keeps its retry
    /// schedule and no span is read twice; a gap wholly covered is not opened.
    #[tokio::test]
    async fn an_overlapping_fill_gap_starts_after_the_kept_one() {
        tokio::time::pause();
        let btc = InstrumentNameExchange::new("BTCUSDT");
        let at = |secs: i64| Utc.timestamp_millis_opt(secs * 1_000).unwrap();
        let mut unrecovered = UnrecoveredFills::default();
        unrecovered.open(std::slice::from_ref(&btc), at(1_000), at(1_100));
        let (_, first) = unrecovered.due(tokio::time::Instant::now())[0].clone();
        unrecovered.failed(&btc, &first, tokio::time::Instant::now());

        // Disconnected again within the first gap's slack: only the rest is a new gap, due now.
        unrecovered.open(std::slice::from_ref(&btc), at(1_105), at(1_200));
        let due = unrecovered.due(tokio::time::Instant::now());
        assert_eq!(due.len(), 1, "the failed gap waits for its retry");
        assert_eq!(due[0].1.start_ms, first.end_ms + 1);
        assert_eq!(due[0].1.end_ms, (1_200 + GAP_END_SLACK_SECS) * 1_000);

        // A gap wholly inside kept ones adds nothing.
        unrecovered.open(std::slice::from_ref(&btc), at(1_150), at(1_150));
        assert_eq!(unrecovered.due(tokio::time::Instant::now()).len(), 1);
        tokio::time::advance(Duration::from_secs(GAP_RETRY_BASE_SECS)).await;
        assert_eq!(unrecovered.due(tokio::time::Instant::now()).len(), 2);
    }

    /// Executions after a span's end are dropped, and their presence ends the walk.
    #[test]
    fn drop_after_keeps_only_the_span() {
        let execution =
            |id: i64, time: Option<i64>| binance_sdk::spot::rest_api::MyTradesResponseInner {
                id: Some(id),
                time,
                ..Default::default()
            };
        let mut page = vec![
            execution(1, Some(100)),
            execution(2, None),
            execution(3, Some(200)),
            execution(4, Some(201)),
        ];
        assert!(drop_after(&mut page, 200));
        assert_eq!(
            page.iter().map(|e| e.id).collect::<Vec<_>>(),
            [Some(1), Some(2), Some(3)]
        );
        assert!(!drop_after(&mut page, 200));
    }

    #[test]
    fn parse_user_data_frame_splits_responses_events_and_unknown_shapes() {
        assert_eq!(
            parse_user_data_frame(r#"{"id":"abc","status":200,"result":{"subscriptionId":0}}"#),
            UserDataFrame::Response
        );
        assert_eq!(
            parse_user_data_frame(
                r#"{"subscriptionId":4,"event":{"e":"outboundAccountPosition","E":1}}"#
            ),
            UserDataFrame::Event {
                subscription_id: Some(4),
                event_type: "outboundAccountPosition",
                event: r#"{"e":"outboundAccountPosition","E":1}"#,
            }
        );
        // A frame with both `id` and `event` is a response.
        assert_eq!(
            parse_user_data_frame(r#"{"id":1,"subscriptionId":4,"event":{"e":"x"}}"#),
            UserDataFrame::Response
        );
        for frame in [
            // A bare event: the shape the spot converter used to assume.
            r#"{"e":"executionReport","i":1}"#,
            // An envelope whose event has no `e`, a non-string `e`, or an escaped one.
            r#"{"subscriptionId":4,"event":{"E":1}}"#,
            r#"{"subscriptionId":4,"event":{"e":7}}"#,
            r#"{"subscriptionId":4,"event":{"e":"executionRep\u006frt"}}"#,
            "not json",
            "[]",
        ] {
            assert_eq!(
                parse_user_data_frame(frame),
                UserDataFrame::Unrecognised,
                "{frame}"
            );
        }
    }

    #[test]
    fn frame_excerpt_cuts_at_200_characters_on_a_char_boundary() {
        assert_eq!(frame_excerpt("short"), "short");
        let exact = "a".repeat(200);
        assert_eq!(frame_excerpt(&exact), exact);
        let long = "é".repeat(300);
        assert_eq!(frame_excerpt(&long).chars().count(), 200);
    }

    /// A `myTrades` execution reduced to the fields recovery's order totals read.
    #[derive(Debug, Clone, Copy)]
    struct Execution {
        id: Option<i64>,
        order_id: Option<i64>,
        qty: Option<&'static str>,
    }

    impl BinanceExecutionFields for Execution {
        fn id(&self) -> Option<i64> {
            self.id
        }
        fn order_id(&self) -> Option<i64> {
            self.order_id
        }
        fn qty(&self) -> Option<&str> {
            self.qty
        }
        // Recovery's order totals never read the time.
        fn time(&self) -> Option<i64> {
            None
        }
    }

    fn execution(id: i64, order_id: i64, qty: &'static str) -> Execution {
        Execution {
            id: Some(id),
            order_id: Some(order_id),
            qty: Some(qty),
        }
    }

    fn btcusdt() -> InstrumentNameExchange {
        InstrumentNameExchange::new("BTCUSDT")
    }

    #[test]
    fn running_totals_accumulate_in_trade_id_order_counting_a_repeat_once() {
        // Out of order, and trade 12 twice, as an overlapping page would serve it.
        let executions = [
            execution(12, 7, "0.5"),
            execution(10, 7, "1"),
            execution(12, 7, "0.5"),
            execution(15, 7, "2"),
        ];

        assert_eq!(
            order_running_totals(7, &executions),
            Some(vec![
                (10, Decimal::ONE),
                (12, Decimal::new(15, 1)),
                (15, Decimal::new(35, 1)),
            ])
        );
    }

    #[test]
    fn running_totals_refuse_a_walk_holding_another_orders_execution() {
        // The venue ignoring `orderId` would serve the symbol's other orders; summing them would
        // report a quantity this order never filled.
        let executions = [execution(10, 7, "1"), execution(11, 8, "5")];

        assert_eq!(order_running_totals(7, &executions), None);
    }

    #[test]
    fn running_totals_refuse_a_walk_with_an_unreadable_execution() {
        let unsized_execution = Execution {
            qty: None,
            ..execution(11, 7, "0")
        };
        let unparseable = execution(11, 7, "abc");
        let unidentified = Execution {
            id: None,
            ..execution(11, 7, "1")
        };

        for broken in [unsized_execution, unparseable, unidentified] {
            assert_eq!(
                order_running_totals(7, &[execution(10, 7, "1"), broken]),
                None,
                "{broken:?}",
            );
        }
    }

    #[tokio::test]
    async fn a_recovered_fill_counts_what_its_order_filled_before_the_window() {
        // Only trade 12 fell inside the recovery window; trade 10 filled before the disconnect.
        let recovered = [execution(12, 7, "2")];
        let order_walk = vec![execution(10, 7, "1"), execution(12, 7, "2")];

        let totals = recovered_order_totals(
            ExchangeId::BinanceSpot,
            &btcusdt(),
            &recovered,
            tokio::time::Instant::now() + Duration::from_secs(5),
            |order_id| {
                assert_eq!(order_id, 7);
                let walk = order_walk.clone();
                async move { Ok(walk) }
            },
        )
        .await;

        assert_eq!(totals.get(&12), Some(&Decimal::new(3, 0)));
    }

    #[tokio::test]
    async fn an_order_whose_walk_fails_is_left_out_and_the_others_are_not() {
        let recovered = [execution(20, 1, "1"), execution(21, 2, "1")];

        let totals = recovered_order_totals(
            ExchangeId::BinanceMargin,
            &btcusdt(),
            &recovered,
            tokio::time::Instant::now() + Duration::from_secs(5),
            |order_id| async move {
                match order_id {
                    1 => Err(UnindexedClientError::Connectivity(
                        ConnectivityError::Socket("connection reset".into()),
                    )),
                    _ => Ok(vec![execution(21, 2, "1")]),
                }
            },
        )
        .await;

        assert_eq!(totals.get(&20), None);
        assert_eq!(totals.get(&21), Some(&Decimal::ONE));
    }

    /// Past the deadline, as when the walks before them used the whole budget, the reads for
    /// client order ids send nothing.
    #[tokio::test]
    async fn a_lookup_past_its_deadline_sends_nothing() {
        let recovered = [execution(20, 1, "1")];
        let cids = recovered_order_cids(
            ExchangeId::BinanceSpot,
            &btcusdt(),
            &recovered,
            tokio::time::Instant::now(),
            |_| async { panic!("no read past the deadline") },
        )
        .await;
        assert!(cids.is_empty());
    }

    /// Each recovered order is read once for its client order id. One the venue reports none for,
    /// and one whose read fails, are left out without holding back the others.
    #[tokio::test]
    async fn recovered_orders_are_read_for_their_client_order_ids() {
        let recovered = [
            execution(20, 1, "1"),
            execution(21, 1, "1"),
            execution(22, 2, "1"),
            execution(23, 3, "1"),
        ];
        let reads = std::sync::atomic::AtomicUsize::new(0);

        let cids = recovered_order_cids(
            ExchangeId::BinanceMargin,
            &btcusdt(),
            &recovered,
            tokio::time::Instant::now() + Duration::from_secs(5),
            |order_id| {
                reads.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                async move {
                    match order_id {
                        1 => Ok(Some(ClientOrderId::new("sweep-1"))),
                        2 => Ok(None),
                        _ => Err(UnindexedClientError::Connectivity(
                            ConnectivityError::Socket("connection reset".into()),
                        )),
                    }
                }
            },
        )
        .await;

        assert_eq!(reads.into_inner(), 3, "one read per order");
        assert_eq!(cids.len(), 1, "{cids:?}");
        assert_eq!(cids.get(&1), Some(&ClientOrderId::new("sweep-1")));
    }

    #[tokio::test(start_paused = true)]
    async fn lookups_stop_at_the_deadline_keeping_the_walks_that_answered() {
        // Order 1's walk never answers; order 2's does. Recovery must return what it has at the
        // deadline, so the fills still go out inside the overall recovery timeout.
        let recovered = [execution(20, 1, "1"), execution(21, 2, "1")];
        let started = tokio::time::Instant::now();
        let deadline = started + Duration::from_secs(5);

        let totals = recovered_order_totals(
            ExchangeId::BinanceSpot,
            &btcusdt(),
            &recovered,
            deadline,
            |order_id| async move {
                if order_id == 1 {
                    std::future::pending::<()>().await;
                }
                Ok(vec![execution(21, 2, "1")])
            },
        )
        .await;

        assert_eq!(totals.get(&20), None);
        assert_eq!(totals.get(&21), Some(&Decimal::ONE));
        assert_eq!(tokio::time::Instant::now(), deadline);
    }

    #[tokio::test(start_paused = true)]
    async fn lookups_run_side_by_side_up_to_the_in_flight_bound() {
        // Each walk takes 4 s against a 5 s budget: one at a time, only the first would finish.
        // Side by side, the first ORDER_EXECUTIONS_IN_FLIGHT finish together at 4 s, and the one
        // over the bound starts only then, so the deadline cuts it.
        let recovered: Vec<Execution> = (1_i64..)
            .take(ORDER_EXECUTIONS_IN_FLIGHT + 1)
            .map(|order_id| execution(100 + order_id, order_id, "1"))
            .collect();
        let over_bound = recovered.last().and_then(|t| t.id);

        let totals = recovered_order_totals(
            ExchangeId::BinanceSpot,
            &btcusdt(),
            &recovered,
            tokio::time::Instant::now() + Duration::from_secs(5),
            |order_id| async move {
                tokio::time::sleep(Duration::from_secs(4)).await;
                Ok(vec![execution(100 + order_id, order_id, "1")])
            },
        )
        .await;

        assert_eq!(totals.len(), ORDER_EXECUTIONS_IN_FLIGHT, "{totals:?}");
        assert!(
            over_bound.is_some_and(|id| !totals.contains_key(&id)),
            "{totals:?}"
        );
    }

    fn classify(
        msg: &str,
        code: Option<i64>,
    ) -> OrderError<AssetNameExchange, InstrumentNameExchange> {
        let err = anyhow::Error::new(ConnectorError::ConnectorClientError {
            msg: msg.to_string(),
            code,
        });
        classify_rest_order_error(&err, &InstrumentNameExchange::new("BTCUSDT"))
    }

    #[test]
    fn codeless_transport_failures_map_to_connectivity() {
        // The SDK's codeless transport/decode wrappers (http_request error arm) must classify as
        // Connectivity — venue status unknown — never as a definitive rejection. Misreporting one
        // of these as Rejected risks a phantom position when the order actually reached the engine.
        // This test also pins the brittle prefix list against silent SDK message-format drift.
        for msg in [
            "HTTP request failed: connection reset",
            "Failed to get response bytes: error reading body",
            "Failed to decompress gzip response",
            "Failed to convert response to UTF-8: invalid utf-8 sequence",
        ] {
            assert!(
                matches!(classify(msg, None), OrderError::Connectivity(_)),
                "expected Connectivity for {msg:?}"
            );
        }
    }

    #[test]
    fn coded_error_maps_to_rejection() {
        // A genuine matching-engine rejection carries a Binance numeric code.
        assert!(matches!(
            classify("Account has insufficient balance.", Some(-2010)),
            OrderError::Rejected(_)
        ));
    }

    #[test]
    fn codeless_non_transport_status_error_maps_to_rejection() {
        // An unusual HTTP error *status* with no Binance code (SDK's catch-all `_` arm) did reach
        // the venue, so it must remain a rejection — not be swallowed as connectivity by an
        // over-broad `code.is_none()` guard. This is the distinction the prefix match preserves.
        assert!(matches!(
            classify("Conflict", None),
            OrderError::Rejected(_)
        ));
    }

    fn classify_err(ce: ConnectorError) -> OrderError<AssetNameExchange, InstrumentNameExchange> {
        classify_rest_order_error(
            &anyhow::Error::new(ce),
            &InstrumentNameExchange::new("BTCUSDT"),
        )
    }

    #[test]
    fn rate_limit_variants_map_to_rejection_ratelimit() {
        // 429/418 surface as RateLimit so callers can back off; venue did respond.
        for ce in [
            ConnectorError::TooManyRequestsError {
                msg: "Too many requests.".to_string(),
                code: Some(-1003),
                retry_after: None,
            },
            ConnectorError::RateLimitBanError {
                msg: "IP banned.".to_string(),
                code: Some(-1003),
                retry_after: None,
            },
        ] {
            assert!(matches!(
                classify_err(ce),
                OrderError::Rejected(ApiError::RateLimit { .. })
            ));
        }
    }

    #[test]
    fn auth_variants_map_to_unauthenticated() {
        // A 401, or a 403 that is an auth failure, is a definitive auth rejection.
        for ce in [
            ConnectorError::UnauthorizedError {
                msg: "bad key".to_string(),
                code: Some(-2014),
            },
            ConnectorError::ForbiddenError {
                msg: "Invalid API-key, IP, or permissions for action.".to_string(),
                code: Some(-2015),
            },
        ] {
            assert!(matches!(
                classify_err(ce),
                OrderError::Rejected(ApiError::Unauthenticated(_))
            ));
        }
    }

    #[test]
    fn server_and_network_variants_map_to_connectivity() {
        // 5xx / transport failures leave the order's venue status unknown → Connectivity.
        for ce in [
            ConnectorError::ServerError {
                msg: "internal error".to_string(),
                status_code: Some(503),
            },
            ConnectorError::NetworkError("connection reset".to_string()),
        ] {
            assert!(matches!(classify_err(ce), OrderError::Connectivity(_)));
        }
    }

    // --- REST query-error classification ---

    fn classify_query(ce: ConnectorError) -> UnindexedClientError {
        classify_rest_query_error(
            &anyhow::Error::new(ce),
            Some(&InstrumentNameExchange::new("BTCUSDT")),
        )
    }

    fn bad_request(code: i64, msg: &str) -> ConnectorError {
        ConnectorError::BadRequestError {
            msg: msg.to_string(),
            code: Some(code),
        }
    }

    #[test]
    fn a_query_the_venue_refuses_is_a_request_rejection_carrying_its_code() {
        let err = classify_query(bad_request(
            -1127,
            "More than 24 hours between startTime and endTime.",
        ));
        let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
            panic!("expected RequestRejected, got {err:?}");
        };
        assert!(msg.contains("-1127"), "code missing from {msg:?}");
        assert!(!err.is_transient());
    }

    #[test]
    fn a_rejected_query_naming_one_instrument_says_which() {
        let err = classify_query(bad_request(
            -11001,
            "Isolated margin account does not exist.",
        ));
        let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
            panic!("expected RequestRejected, got {err:?}");
        };
        assert!(msg.starts_with("BTCUSDT: "), "symbol missing from {msg:?}");
        assert!(msg.contains("-11001"), "code missing from {msg:?}");

        // With no single instrument in the request there is none to name.
        let err = classify_rest_query_error(
            &anyhow::Error::new(bad_request(-1127, "More than 24 hours.")),
            None,
        );
        let UnindexedClientError::Api(ApiError::RequestRejected(msg)) = &err else {
            panic!("expected RequestRejected, got {err:?}");
        };
        assert!(!msg.contains("BTCUSDT"), "{msg:?}");
    }

    #[test]
    fn a_query_naming_an_invalid_symbol_is_an_invalid_instrument() {
        let err = classify_query(bad_request(-1121, "Invalid symbol."));
        assert!(
            matches!(
                &err,
                UnindexedClientError::Api(ApiError::InstrumentInvalid(instrument, _))
                    if instrument.name().as_str() == "BTCUSDT"
            ),
            "got {err:?}"
        );

        // With no single instrument in the request there is none to name.
        let err = classify_rest_query_error(
            &anyhow::Error::new(bad_request(-1121, "Invalid symbol.")),
            None,
        );
        assert!(
            matches!(err, UnindexedClientError::Api(ApiError::RequestRejected(_))),
            "got {err:?}"
        );
    }

    #[test]
    fn order_codes_do_not_leak_into_query_classification() {
        // On an order path -2011/-2013 mean "already cancelled"; a query has no order to cancel.
        for code in [-2011, -2013] {
            let err = classify_query(bad_request(code, "Unknown order sent."));
            assert!(
                matches!(err, UnindexedClientError::Api(ApiError::RequestRejected(_))),
                "{code}: got {err:?}"
            );
        }
    }

    #[test]
    fn query_rejections_with_auth_or_throttle_codes_keep_their_meaning() {
        for (code, msg) in [
            (-1002, "You are not authorized to execute this request."),
            (-2015, "Invalid API-key, IP, or permissions for action."),
            (-1022, "Signature for this request is not valid."),
            (-2014, "API-key format invalid."),
        ] {
            let err = classify_query(bad_request(code, msg));
            assert!(
                matches!(err, UnindexedClientError::Api(ApiError::Unauthenticated(_))),
                "{code}: got {err:?}"
            );
        }
        for code in [-1003, -1015] {
            let err = classify_query(bad_request(code, "Too much request weight used."));
            assert!(
                matches!(err, UnindexedClientError::Api(ApiError::RateLimit { .. })),
                "{code}: got {err:?}"
            );
        }
    }

    #[test]
    fn query_rejections_a_retry_can_cure_stay_transient() {
        for code in [-1000, -1001, -1006, -1007, -1008, -1021] {
            let err = classify_query(bad_request(code, "Internal error; please try again."));
            assert!(
                matches!(err, UnindexedClientError::Connectivity(_)),
                "{code}: got {err:?}"
            );
            assert!(err.is_transient());
        }
    }

    #[test]
    fn query_transport_server_and_throttle_failures_classify_like_orders() {
        for ce in [
            ConnectorError::ConnectorClientError {
                msg: "HTTP request failed: connection reset".to_string(),
                code: None,
            },
            ConnectorError::ServerError {
                msg: "Server error: 503".to_string(),
                status_code: Some(503),
            },
            ConnectorError::NetworkError("connection reset".to_string()),
        ] {
            let err = classify_query(ce);
            assert!(
                matches!(err, UnindexedClientError::Connectivity(_)),
                "got {err:?}"
            );
            assert!(err.is_transient());
        }

        let err = classify_query(ConnectorError::TooManyRequestsError {
            msg: "Too many requests.".to_string(),
            code: Some(-1003),
            retry_after: None,
        });
        assert!(matches!(
            err,
            UnindexedClientError::Api(ApiError::RateLimit { .. })
        ));

        let err = classify_query(ConnectorError::UnauthorizedError {
            msg: "API-key format invalid.".to_string(),
            code: Some(-2014),
        });
        let UnindexedClientError::Api(ApiError::Unauthenticated(msg)) = &err else {
            panic!("expected Unauthenticated, got {err:?}");
        };
        assert!(msg.contains("-2014"), "code missing from {msg:?}");
    }

    #[test]
    fn a_request_the_client_failed_to_build_is_internal() {
        let err = classify_rest_query_error(
            &anyhow::Error::new(ParamBuildError::UninitializedField("symbol")),
            None,
        );
        assert!(
            matches!(&err, UnindexedClientError::Internal(msg) if msg.contains("symbol")),
            "got {err:?}"
        );
        assert!(!err.is_transient());
    }

    #[test]
    fn a_response_body_that_does_not_decode_is_internal() {
        // What binance-sdk's `data()` returns when the body does not fit its model: the
        // `serde_json` error as a codeless `ConnectorClientError`.
        let err = response_decode_error(ConnectorError::ConnectorClientError {
            msg: "invalid type: map, expected a sequence at line 1 column 0".to_string(),
            code: None,
        });
        assert!(
            matches!(&err, UnindexedClientError::Internal(msg) if msg.contains("invalid type")),
            "got {err:?}"
        );
        assert!(!err.is_transient());
    }

    #[test]
    fn untyped_query_errors_are_classified_by_their_text() {
        // Errors from the SDK's request plumbing arrive as untyped text, not a ConnectorError.
        for msg in [
            "Error -1002: unauthorized",
            // Code only, no auth wording: isolates the numeric-code branch.
            "Error -2015: permission denied for action",
            "invalid signature provided",
            "The signature for this request is not valid.",
            "Invalid API-key format",
        ] {
            let err = classify_rest_query_error(&anyhow::anyhow!("{msg}"), None);
            assert!(
                matches!(err, UnindexedClientError::Api(ApiError::Unauthenticated(_))),
                "{msg:?}: got {err:?}"
            );
        }

        // Anything else failed before the request left, and fails the same way on retry.
        let err = classify_rest_query_error(
            &anyhow::anyhow!("relative URL without a base").context("Failed to join base URL"),
            None,
        );
        assert!(
            matches!(err, UnindexedClientError::Internal(_)),
            "got {err:?}"
        );
        assert!(!err.is_transient());
    }

    #[test]
    fn a_forbidden_query_is_an_auth_failure_only_when_it_says_so() {
        // Binance documents 403 as its WAF limit: back off and retry.
        let err = classify_query(ConnectorError::ForbiddenError {
            msg: "<html>403 Forbidden</html>".to_string(),
            code: None,
        });
        assert!(
            matches!(err, UnindexedClientError::Api(ApiError::RateLimit { .. })),
            "got {err:?}"
        );

        let err = classify_query(ConnectorError::ForbiddenError {
            msg: "Invalid API-key, IP, or permissions for action.".to_string(),
            code: Some(-2015),
        });
        assert!(
            matches!(err, UnindexedClientError::Api(ApiError::Unauthenticated(_))),
            "got {err:?}"
        );
    }

    #[test]
    fn query_classification_reads_the_sdk_error_through_added_context() {
        let err = classify_rest_query_error(
            &anyhow::Error::new(bad_request(-1127, "More than 24 hours.")).context("outer"),
            None,
        );
        assert!(
            matches!(err, UnindexedClientError::Api(ApiError::RequestRejected(_))),
            "got {err:?}"
        );
    }

    // --- Order rejection codes ---

    #[test]
    fn an_order_forbidden_by_the_waf_is_a_rate_limit() {
        // Refused at the edge: the order never reached the matching engine.
        assert!(matches!(
            classify_err(ConnectorError::ForbiddenError {
                msg: "<html>403 Forbidden</html>".to_string(),
                code: None,
            }),
            OrderError::Rejected(ApiError::RateLimit { .. })
        ));
    }

    #[test]
    fn an_order_the_venue_failed_has_an_unknown_status() {
        // Binance: the order may have executed. A definitive rejection would hide a fill.
        for code in [-1000, -1001, -1006, -1007, -1008] {
            let err = classify_err(ConnectorError::BadRequestError {
                msg: "Timeout waiting for response from backend server.".to_string(),
                code: Some(code),
            });
            assert!(
                matches!(err, OrderError::Connectivity(_)),
                "{code}: got {err:?}"
            );
            assert!(err.is_transient());
        }

        // The WS-API reports the same codes in its `ResponseError` text.
        let err = parse_binance_order_rejection(
            "Server\u{2010}side response error (code -1007): Timeout waiting for response from \
             backend server. Send status unknown; execution status unknown."
                .to_string(),
            &InstrumentNameExchange::new("BTCUSDT"),
        );
        assert!(matches!(err, OrderError::Connectivity(_)), "got {err:?}");
    }

    #[test]
    fn an_order_with_a_bad_signature_or_key_format_is_unauthenticated() {
        for (code, msg) in [
            (-1022, "Signature for this request is not valid."),
            (-2014, "API-key format invalid."),
        ] {
            let err = classify_err(ConnectorError::BadRequestError {
                msg: msg.to_string(),
                code: Some(code),
            });
            assert!(
                matches!(err, OrderError::Rejected(ApiError::Unauthenticated(_))),
                "{code}: got {err:?}"
            );
        }
    }

    #[test]
    fn a_definitive_order_rejection_is_still_a_rejection() {
        let err = classify_err(ConnectorError::BadRequestError {
            msg: "Account has insufficient balance for requested action.".to_string(),
            code: Some(-2010),
        });
        assert!(
            matches!(
                err,
                OrderError::Rejected(ApiError::BalanceInsufficient(None, _))
            ),
            "got {err:?}"
        );
    }

    /// `-2010` is Binance's generic NEW_ORDER_REJECTED, so only its text says whether the balance
    /// ran short, and Binance never names the asset. Margin reaches the venue over REST, Spot over
    /// the WS API; both must read it the same way.
    #[test]
    fn a_2010_rejection_is_read_by_its_message() {
        let cases: [(&str, fn(&UnindexedOrderError) -> bool); 4] = [
            (
                "Account has insufficient balance for requested action.",
                |err| {
                    matches!(
                        err,
                        OrderError::Rejected(ApiError::BalanceInsufficient(None, _))
                    )
                },
            ),
            ("Order would trigger immediately.", |err| {
                matches!(err, OrderError::Rejected(ApiError::OrderRejected(_)))
            }),
            // The `clOrdId` is already in use.
            ("Duplicate order sent.", |err| {
                matches!(
                    err,
                    OrderError::Rejected(ApiError::DuplicateClientOrderId(_))
                )
            }),
            (
                "Trailing stop orders are not supported for this symbol.",
                |err| matches!(err, OrderError::Rejected(ApiError::OrderRejected(_))),
            ),
        ];
        for (msg, expected) in cases {
            let rest = classify_err(ConnectorError::BadRequestError {
                msg: msg.to_string(),
                code: Some(-2010),
            });
            assert!(expected(&rest), "REST {msg:?}: got {rest:?}");

            let ws = classify_ws(-2010, msg);
            assert!(
                ws.as_ref().is_some_and(expected),
                "WS API {msg:?}: got {ws:?}"
            );
        }
    }

    /// A margin order that needs a borrow is refused by a `-3xxx` code. A refused borrow is
    /// [`ApiError::BorrowRejected`] by its code, over REST (the code spliced into the message) and
    /// over the WS API (the code in the error's text) alike, whatever the wording says.
    #[test]
    fn a_borrow_refusal_is_read_by_its_code() {
        for (code, msg, reason) in [
            (
                -3006,
                "Your borrow amount has exceed maximum borrow amount.",
                BorrowRejectReason::BorrowLimitExceeded,
            ),
            (
                -3008,
                "Borrow not allowed.",
                BorrowRejectReason::BorrowDisabled,
            ),
            (
                -3012,
                "Borrow is banned for this asset.",
                BorrowRejectReason::BorrowDisabled,
            ),
            (
                -3014,
                "Borrow is banned for this account.",
                BorrowRejectReason::BorrowDisabled,
            ),
            (
                -3045,
                "The system doesn't have enough asset now.",
                BorrowRejectReason::InventoryUnavailable,
            ),
        ] {
            let rest = classify_err(ConnectorError::BadRequestError {
                msg: msg.to_string(),
                code: Some(code),
            });
            let expected =
                BorrowReject::new(reason, Some(code.to_string()), format!("{code} {msg}"));
            assert_eq!(
                rest,
                OrderError::Rejected(ApiError::BorrowRejected(expected)),
                "REST {code}"
            );

            let ws = classify_ws(code, msg);
            assert!(
                matches!(
                    &ws,
                    Some(OrderError::Rejected(ApiError::BorrowRejected(reject)))
                        if reject.reason == reason
                            && reject.venue_code.as_deref() == Some(code.to_string().as_str())
                ),
                "WS API {code}: got {ws:?}"
            );
        }
    }

    /// A cross margin account refuses a trailing `STOP_LOSS` with `-3111`. The venue does not
    /// take that order type, so it is [`OrderError::UnsupportedOrderType`] carrying the venue's
    /// message, over REST (the code spliced into the message) and over the WS API alike, and a
    /// retry cannot cure it.
    #[test]
    fn a_trailing_stop_limit_only_refusal_is_an_unsupported_order_type() {
        let msg = "Trailing stop order only supports limit type.";
        let rest = classify_err(ConnectorError::BadRequestError {
            msg: msg.to_string(),
            code: Some(-3111),
        });
        assert_eq!(
            rest,
            OrderError::UnsupportedOrderType(format!("-3111 {msg}"))
        );
        assert!(!rest.is_transient());

        let ws = classify_ws(-3111, msg);
        assert!(
            matches!(
                &ws,
                Some(OrderError::UnsupportedOrderType(text))
                    if text.contains("-3111") && text.contains(msg)
            ),
            "WS API: got {ws:?}"
        );
    }

    /// The margin account's own collateral running short is a shortfall, not a refused borrow,
    /// and goes by code. A borrow below the minimum, or a pair or account barred from trading, is
    /// a plain rejection.
    #[test]
    fn other_margin_codes_are_not_borrow_refusals() {
        let instrument = InstrumentNameExchange::new("BTCUSDT");
        for msg in [
            "-3041 Balance is not enough",
            "-3023 You can't transfer out/place order under current margin level.",
        ] {
            assert_eq!(
                parse_binance_api_error(msg.to_string(), &instrument),
                ApiError::BalanceInsufficient(None, msg.to_string()),
                "{msg:?}"
            );
        }
        for msg in [
            "-3013 Borrow amount less than minimum borrow amount.",
            "-3021 Margin account are not allowed to trade this trading pair.",
            "-3022 You account's trading is banned.",
        ] {
            assert_eq!(
                parse_binance_api_error(msg.to_string(), &instrument),
                ApiError::OrderRejected(msg.to_string()),
                "{msg:?}"
            );
        }
    }

    #[test]
    fn order_rejections_keep_their_order_meaning() {
        let instrument = InstrumentNameExchange::new("BTCUSDT");
        for (msg, expected) in [
            // Cancel races stay "already cancelled".
            (
                "-2013 Order does not exist.",
                ApiError::OrderAlreadyCancelled,
            ),
            ("-2011 Unknown order sent.", ApiError::OrderAlreadyCancelled),
            // A stale timestamp on an order is a definitive refusal: it did not execute.
            (
                "-1021 Timestamp for this request is outside of the recvWindow.",
                ApiError::OrderRejected(
                    "-1021 Timestamp for this request is outside of the recvWindow.".to_string(),
                ),
            ),
            // A duplicated parameter is a malformed request, not a client order id in use.
            (
                "-1102 Duplicate values for a parameter detected.",
                ApiError::OrderRejected(
                    "-1102 Duplicate values for a parameter detected.".to_string(),
                ),
            ),
            // Auth by wording alone, with no code.
            (
                "Invalid API-key, IP, or permissions for action.",
                ApiError::Unauthenticated(
                    "Invalid API-key, IP, or permissions for action.".to_string(),
                ),
            ),
        ] {
            assert_eq!(
                parse_binance_order_rejection(msg.to_string(), &instrument),
                OrderError::Rejected(expected),
                "{msg:?}"
            );
        }
    }

    fn classify_ws(code: i64, message: &str) -> Option<UnindexedOrderError> {
        classify_ws_order_error(
            &anyhow::Error::new(WebsocketError::ResponseError {
                code,
                message: message.to_string(),
                retry_after: None,
            }),
            &InstrumentNameExchange::new("BTCUSDT"),
        )
    }

    #[test]
    fn a_dropped_connection_names_its_cause() {
        assert_eq!(disconnect_error(true), "heartbeat timeout (30s)");
        assert_eq!(
            disconnect_error(false),
            "WebSocket disconnected (server close/error)"
        );
    }

    /// A reconnect facing a cooldown longer than a request sleeps through ends the stream at once,
    /// spending no attempt and no time, and so does a connect that finds it.
    #[tokio::test(start_paused = true)]
    async fn a_reconnect_ends_at_once_on_a_cooldown_too_long_to_wait_through() {
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.on_rate_limited(Some(Duration::from_secs(60)));
        let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
        let started = tokio::time::Instant::now();
        let ended = StreamTerminationReason::RateLimited {
            retry_after: Duration::from_secs(60),
        };

        let reason = retry_or_end(&mut budget, &tracker, || {
            unreachable!("the budget is not spent")
        })
        .await;
        assert_eq!(reason, Err(ended.clone()));
        assert_eq!(tracker.before_stream_connect().await, Err(ended));
        assert_eq!(started.elapsed(), Duration::ZERO);
        assert_eq!(budget.attempts(), 0);
    }

    /// A cooldown short enough to sleep through is waited out before the connect, and that wait
    /// spends no attempt: only the failure that preceded it does.
    #[tokio::test(start_paused = true)]
    async fn a_short_cooldown_is_waited_out_without_spending_an_attempt() {
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.on_rate_limited(Some(Duration::from_secs(5)));
        let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
        let started = tokio::time::Instant::now();

        assert_eq!(
            retry_or_end(&mut budget, &tracker, String::new).await,
            Ok(())
        );
        assert_eq!(started.elapsed(), BackoffLimits::ACCOUNT_STREAM.initial);
        assert_eq!(tracker.before_stream_connect().await, Ok(()));
        assert_eq!(started.elapsed(), Duration::from_secs(5));
        assert_eq!(budget.attempts(), 1);
    }

    /// A refusal that names no wait starts the default cooldown, which is waited out, so
    /// refusals that keep coming spend the budget and end the stream on it, not on the cooldown.
    #[tokio::test(start_paused = true)]
    async fn refusals_naming_no_wait_spend_the_budget() {
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        let mut budget = ReconnectBudget::new("Binance", BackoffLimits::ACCOUNT_STREAM);
        let reason = loop {
            assert_eq!(tracker.before_stream_connect().await, Ok(()));
            tracker.on_rate_limited(None);
            if let Err(reason) = retry_or_end(&mut budget, &tracker, || "refused".to_string()).await
            {
                break reason;
            }
        };

        assert_eq!(
            reason,
            StreamTerminationReason::ReconnectBudgetExhausted {
                attempts: BackoffLimits::ACCOUNT_STREAM.max_attempts,
                last_error: "refused".to_string(),
            }
        );
    }

    /// A refused handshake and a WS-API rate-limit response start a cooldown: a 429 the default,
    /// a 418 Binance's shortest ban, and a response as long as the ban it names. Any other failure
    /// starts none.
    #[tokio::test]
    async fn only_a_rate_limit_refusal_of_a_stream_connect_starts_a_cooldown() {
        let handshake = |msg: &str| anyhow::Error::new(WebsocketError::Handshake(msg.into()));
        let response = |code, retry_after| {
            anyhow::Error::new(WebsocketError::ResponseError {
                code,
                message: "refused".to_string(),
                retry_after,
            })
            .context("subscribing")
        };
        let about = |tracker: &RateLimitTracker, wait: Duration| {
            tracker
                .cooldown_remaining()
                .is_some_and(|left| left <= wait && left > wait - Duration::from_secs(5))
        };

        let tracker = RateLimitTracker::new(WeightPool::Spot);
        assert!(!tracker.on_stream_refused(&handshake("HTTP error: 503 Service Unavailable")));
        let unrelated = response(-2010, None);
        assert!(!tracker.on_stream_refused(&unrelated));
        assert!(!tracker.on_stream_refused(&anyhow::anyhow!("connection reset")));
        assert_eq!(tracker.cooldown_remaining(), None);

        assert!(tracker.on_stream_refused(&handshake("HTTP error: 429 Too Many Requests")));
        let default = Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS);
        assert!(
            about(&tracker, default),
            "{:?}",
            tracker.cooldown_remaining()
        );

        let tracker = RateLimitTracker::new(WeightPool::Spot);
        assert!(tracker.on_stream_refused(&handshake("HTTP error: 418 I'm a teapot")));
        assert!(
            about(&tracker, MIN_IP_BAN),
            "{:?}",
            tracker.cooldown_remaining()
        );

        let tracker = RateLimitTracker::new(WeightPool::Spot);
        let until = i64::try_from(unix_ms()).unwrap() + 120_000;
        assert!(tracker.on_stream_refused(&response(-1003, Some(until))));
        let ban = Duration::from_millis(120_000 + CLOCK_SKEW_SLACK_MS);
        assert!(about(&tracker, ban), "{:?}", tracker.cooldown_remaining());
    }

    /// Opening a stream during a long cooldown fails unsent with the time left, and a refusal is
    /// reported with the cooldown it started; any other failure as the caller makes it.
    #[tokio::test]
    async fn opening_a_stream_reports_a_rate_limit_with_the_cooldown_left() {
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        let socket = |e: &anyhow::Error| {
            UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
        };
        let other = anyhow::anyhow!("connection reset");
        assert_eq!(tracker.stream_open_error(&other, socket), socket(&other));
        assert_eq!(tracker.before_stream_open().await, Ok(()));

        let refused = anyhow::Error::new(WebsocketError::Handshake(
            "HTTP error: 429 Too Many Requests".into(),
        ));
        let error = tracker.stream_open_error(&refused, socket);
        assert!(
            error
                .retry_after()
                .is_some_and(|wait| wait > Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS - 5)),
            "{error:?}"
        );

        tracker.on_rate_limited(Some(Duration::from_secs(60)));
        let error = tracker.before_stream_open().await.unwrap_err();
        assert!(
            error
                .retry_after()
                .is_some_and(|wait| wait > Duration::from_secs(55)),
            "{error:?}"
        );
    }

    #[test]
    fn a_ws_response_without_an_error_body_is_read_by_its_status() {
        // With no error object the SDK reports the HTTP status as `code`, "Unknown error" as text.
        let err = classify_ws(503, "Unknown error");
        assert!(
            matches!(err, Some(OrderError::Connectivity(_))),
            "a 5xx leaves the order's status unknown: {err:?}"
        );
        for code in [429, 418, 403] {
            let err = classify_ws(code, "Unknown error");
            assert!(
                matches!(err, Some(OrderError::Rejected(ApiError::RateLimit { .. }))),
                "{code}: got {err:?}"
            );
        }
        let err = classify_ws(401, "Unknown error");
        assert!(
            matches!(
                err,
                Some(OrderError::Rejected(ApiError::Unauthenticated(_)))
            ),
            "got {err:?}"
        );
        let err = classify_ws(409, "Unknown error");
        assert!(
            matches!(err, Some(OrderError::Rejected(ApiError::OrderRejected(_)))),
            "got {err:?}"
        );
    }

    #[test]
    fn a_ws_response_with_a_binance_code_is_read_by_its_code() {
        let err = classify_ws(-1007, "Timeout waiting for response from backend server.");
        assert!(
            matches!(err, Some(OrderError::Connectivity(_))),
            "got {err:?}"
        );

        // A context layer must not hide the code.
        let wrapped = anyhow::Error::new(WebsocketError::ResponseError {
            code: -1007,
            message: "Timeout waiting for response from backend server.".to_string(),
            retry_after: None,
        })
        .context("outer");
        let err = classify_ws_order_error(&wrapped, &InstrumentNameExchange::new("BTCUSDT"));
        assert!(
            matches!(err, Some(OrderError::Connectivity(_))),
            "got {err:?}"
        );
        let err = classify_ws(-1003, "Too much request weight used.");
        assert!(
            matches!(err, Some(OrderError::Rejected(ApiError::RateLimit { .. }))),
            "got {err:?}"
        );
    }

    #[test]
    fn a_waf_403_is_backed_off_as_a_rate_limit_but_an_auth_403_is_not() {
        let waf = anyhow::Error::new(ConnectorError::ForbiddenError {
            msg: "<html>403 Forbidden</html>".to_string(),
            code: None,
        });
        assert!(is_rate_limit_error(&waf));

        let auth = anyhow::Error::new(ConnectorError::ForbiddenError {
            msg: "Invalid API-key, IP, or permissions for action.".to_string(),
            code: Some(-2015),
        });
        assert!(!is_rate_limit_error(&auth));
    }

    /// The weight-per-minute entry a WS-API response carries.
    fn ws_weight(limit: u32, count: u32) -> Vec<WebsocketApiRateLimit> {
        vec![
            WebsocketApiRateLimit {
                rate_limit_type: RateLimitType::Orders,
                interval: Interval::Second,
                interval_num: 10,
                limit: 100,
                count: 100,
            },
            WebsocketApiRateLimit {
                rate_limit_type: RateLimitType::RequestWeight,
                interval: Interval::Minute,
                interval_num: 1,
                limit,
                count,
            },
        ]
    }

    /// Whether a wait of `kind` is still pending after 1 ms of (paused) time.
    async fn waits(tracker: &RateLimitTracker, kind: RequestKind) -> bool {
        tokio::time::timeout(Duration::from_millis(1), tracker.wait_if_blocked(kind))
            .await
            .is_err()
    }

    /// The pause runs to just past the end of the minute the request was sent in, and a
    /// response observed after that pauses nothing.
    #[test]
    fn pause_after_ends_just_past_the_send_minute() {
        let minute = 29_000_000 * 60_000_u128;
        let slack = CLOCK_SKEW_SLACK_MS;
        let ms = |ms: u64| Some(Duration::from_millis(ms));
        assert_eq!(
            pause_after(minute + 15_000, minute + 15_000),
            ms(45_000 + slack)
        );
        assert_eq!(pause_after(minute, minute + 100), ms(59_900 + slack));
        assert_eq!(pause_after(minute + 59_999, minute + 59_999), ms(1 + slack));
        // Sent in the previous minute, observed during the slack: only the slack's rest.
        assert_eq!(pause_after(minute - 100, minute + 400), ms(slack - 400));
        // Sent in the previous minute, observed after it and the slack ended.
        assert_eq!(pause_after(minute - 100, minute + u128::from(slack)), None);
        assert_eq!(pause_after(minute - 100, minute + 30_000), None);
    }

    /// At the threshold, queries pause and orders do not; one under it, nothing pauses.
    #[tokio::test]
    async fn weight_near_the_limit_pauses_queries_but_not_orders() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.observe_ws_api(Some(&ws_weight(1_000, 899)), unix_ms());
        assert!(!waits(&tracker, RequestKind::Query).await, "below 90%");

        tracker.observe_ws_api(Some(&ws_weight(1_000, 900)), unix_ms());
        assert!(waits(&tracker, RequestKind::Query).await, "at 90%");
        assert!(!waits(&tracker, RequestKind::Order).await);
        assert!(
            !tracker.is_blocked(),
            "a pause is not a rate-limit cooldown"
        );

        tokio::time::advance(Duration::from_millis(60_000 + CLOCK_SKEW_SLACK_MS)).await;
        assert!(!waits(&tracker, RequestKind::Query).await, "the pause ends");
    }

    /// Essential reads skip the pause near the limit, like orders, but wait out a cooldown.
    #[tokio::test]
    async fn essential_reads_skip_the_pause_but_not_a_cooldown() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.throttle(Duration::from_secs(30));
        assert!(waits(&tracker, RequestKind::Query).await);
        assert!(!waits(&tracker, RequestKind::Essential).await);

        tracker.on_rate_limited(Some(Duration::from_secs(5)));
        assert!(waits(&tracker, RequestKind::Essential).await);
    }

    /// A rate-limit cooldown holds back orders as well as queries.
    #[tokio::test]
    async fn a_rate_limit_cooldown_holds_back_both_kinds() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.on_rate_limited(Some(Duration::from_secs(5)));
        assert!(tracker.is_blocked());
        assert!(waits(&tracker, RequestKind::Order).await);
        assert!(waits(&tracker, RequestKind::Query).await);

        tokio::time::advance(Duration::from_secs(5)).await;
        assert!(!tracker.is_blocked());
        assert!(!waits(&tracker, RequestKind::Order).await);
    }

    /// A cooldown of at most `MAX_COOLDOWN_SLEEP` is slept through; one longer fails at once,
    /// for every kind, with the time left.
    #[tokio::test]
    async fn a_cooldown_longer_than_the_sleep_limit_fails_at_once() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.on_rate_limited(Some(MAX_COOLDOWN_SLEEP));
        assert!(waits(&tracker, RequestKind::Order).await, "at the limit");

        let longer = MAX_COOLDOWN_SLEEP + Duration::from_millis(1);
        tracker.on_rate_limited(Some(longer));
        for kind in [
            RequestKind::Order,
            RequestKind::Query,
            RequestKind::Essential,
        ] {
            assert_eq!(
                tracker.wait_if_blocked(kind).now_or_never(),
                Some(Err(CooldownActive { remaining: longer })),
                "{kind:?}"
            );
        }
        assert_eq!(tracker.cooldown_remaining(), Some(longer));
    }

    /// A request sleeping through a short cooldown fails once another refusal extends it past
    /// the limit, rather than sleeping through the longer one.
    #[tokio::test]
    async fn a_cooldown_extended_past_the_limit_while_sleeping_fails() {
        tokio::time::pause();
        let tracker = Arc::new(RateLimitTracker::new(WeightPool::Spot));
        tracker.on_rate_limited(Some(Duration::from_secs(5)));
        let waiter = tokio::spawn({
            let tracker = Arc::clone(&tracker);
            async move { tracker.wait_if_blocked(RequestKind::Order).await }
        });
        tokio::time::advance(Duration::from_secs(1)).await;
        tracker.on_rate_limited(Some(Duration::from_secs(120)));
        tokio::time::advance(Duration::from_secs(4)).await;

        // The timer wakes the sleeper up to a millisecond late.
        let remaining = waiter.await.unwrap().unwrap_err().remaining;
        assert!(
            remaining <= Duration::from_secs(116)
                && remaining >= Duration::from_secs(116) - Duration::from_millis(1),
            "{remaining:?}"
        );
    }

    /// A weight pause longer than the cooldown limit is slept through: it is not a refusal.
    #[tokio::test]
    async fn a_long_weight_pause_is_slept_through() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.throttle(Duration::from_secs(30));
        assert!(waits(&tracker, RequestKind::Query).await);
        tokio::time::advance(Duration::from_secs(30)).await;
        assert_eq!(
            tracker.wait_if_blocked(RequestKind::Query).now_or_never(),
            Some(Ok(()))
        );
    }

    /// A wait beyond the clock's range does not panic, and blocks every request.
    #[tokio::test]
    async fn a_wait_beyond_the_clock_range_blocks_without_panicking() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.on_rate_limited(Some(Duration::MAX));
        assert_eq!(
            tracker.cooldown_remaining(),
            Some(Duration::from_secs(u64::from(u32::MAX)))
        );
        assert!(
            tracker
                .wait_if_blocked(RequestKind::Order)
                .now_or_never()
                .is_some_and(|waited| waited.is_err())
        );
    }

    #[tokio::test]
    async fn cooldown_remaining_reports_none_outside_a_cooldown() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        assert_eq!(tracker.cooldown_remaining(), None);
        tracker.on_rate_limited(None);
        assert_eq!(
            tracker.cooldown_remaining(),
            Some(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS))
        );
        tokio::time::advance(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS)).await;
        assert_eq!(tracker.cooldown_remaining(), None);
    }

    /// The wait runs to the slack past the time Binance named, and is zero once that has passed.
    #[test]
    fn wait_until_epoch_ms_runs_to_the_slack_past_the_time() {
        let now = 1_700_000_000_000_u128;
        let at = |offset: i128| i64::try_from(now.cast_signed() + offset).unwrap();
        let ms = Duration::from_millis;
        assert_eq!(
            wait_until_epoch_ms(at(60_000), now),
            ms(60_000 + CLOCK_SKEW_SLACK_MS)
        );
        assert_eq!(wait_until_epoch_ms(at(0), now), ms(CLOCK_SKEW_SLACK_MS));
        assert_eq!(
            wait_until_epoch_ms(at(-400), now),
            ms(CLOCK_SKEW_SLACK_MS - 400)
        );
        let slack = i128::from(CLOCK_SKEW_SLACK_MS);
        assert_eq!(wait_until_epoch_ms(at(-slack), now), Duration::ZERO);
        assert_eq!(wait_until_epoch_ms(at(-60_000), now), Duration::ZERO);
        assert_eq!(wait_until_epoch_ms(-1, now), Duration::ZERO);
        assert_eq!(
            wait_until_epoch_ms(i64::MAX, 0),
            ms(i64::MAX.cast_unsigned() + CLOCK_SKEW_SLACK_MS)
        );
    }

    /// A WS-API ban epoch `ms` from now.
    fn epoch_in(ms: u64) -> i64 {
        i64::try_from(unix_ms()).unwrap() + i64::try_from(ms).unwrap()
    }

    fn classify_ws_with_retry_after(code: i64, retry_after: Option<i64>) -> UnindexedOrderError {
        classify_ws_order_error(
            &anyhow::Error::new(WebsocketError::ResponseError {
                code,
                message: "Way too much request weight used; IP banned.".to_string(),
                retry_after,
            }),
            &InstrumentNameExchange::new("BTCUSDT"),
        )
        .expect("a ResponseError is a venue response")
    }

    /// The wait a rate-limit refusal reports, or `None` for any other error.
    fn order_retry_after(err: &UnindexedOrderError) -> Option<Option<Duration>> {
        match err {
            OrderError::Rejected(api @ ApiError::RateLimit { .. }) => Some(api.retry_after()),
            _ => None,
        }
    }

    /// A WS-API throttle or ban reports the wait until its `retryAfter`, plus the slack.
    #[test]
    fn a_ws_api_rate_limit_reports_the_wait_until_retry_after() {
        for code in [-1003, -1015, 429, 418] {
            let err = classify_ws_with_retry_after(code, Some(epoch_in(120_000)));
            let wait = order_retry_after(&err)
                .flatten()
                .unwrap_or_else(|| panic!("{code}: got {err:?}"));
            let expected = Duration::from_millis(120_000 + CLOCK_SKEW_SLACK_MS);
            assert!(
                wait <= expected && wait > expected - Duration::from_secs(5),
                "{code}: {wait:?}"
            );

            let err = classify_ws_with_retry_after(code, None);
            assert_eq!(order_retry_after(&err), Some(None), "{code}");

            // A ban that has lifted, slack included, names no wait.
            let past = epoch_in(0) - i64::try_from(CLOCK_SKEW_SLACK_MS).unwrap() - 1_000;
            let err = classify_ws_with_retry_after(code, Some(past));
            assert_eq!(order_retry_after(&err), Some(None), "{code}");
        }
    }

    /// A `retryAfter` on any other refusal does not make it a rate limit.
    #[test]
    fn a_retry_after_on_another_refusal_is_not_a_rate_limit() {
        let err = classify_ws_with_retry_after(-2010, Some(epoch_in(60_000)));
        assert!(
            matches!(err, OrderError::Rejected(ApiError::OrderRejected(_))),
            "got {err:?}"
        );
    }

    /// REST's `Retry-After`, in seconds, is the wait on a 429 and a 418, for orders and queries.
    #[test]
    fn a_rest_rate_limit_reports_its_retry_after() {
        let refusal = |banned: bool, retry_after| {
            let msg = "Too many requests.".to_string();
            let code = Some(-1003);
            if banned {
                ConnectorError::RateLimitBanError {
                    msg,
                    code,
                    retry_after,
                }
            } else {
                ConnectorError::TooManyRequestsError {
                    msg,
                    code,
                    retry_after,
                }
            }
        };
        // A zero wait names none.
        for (sent, expected) in [
            (Some(120), Some(Duration::from_secs(120))),
            (Some(0), None),
            (None, None),
        ] {
            for banned in [false, true] {
                let order = classify_err(refusal(banned, sent));
                assert_eq!(order_retry_after(&order), Some(expected), "{order:?}");
                let query = classify_query(refusal(banned, sent));
                assert!(
                    matches!(
                        &query,
                        UnindexedClientError::Api(api @ ApiError::RateLimit { .. })
                            if api.retry_after() == expected
                    ),
                    "{query:?}"
                );
            }
        }
    }

    /// A throttle without a wait is retried after each backoff until the retries run out. The
    /// last refusal is returned, and holds every request back for the default cooldown.
    #[tokio::test]
    async fn a_throttle_is_retried_until_the_retries_run_out() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        let started = tokio::time::Instant::now();
        let mut calls = 0_u32;
        let result =
            rest_call_with_retry::<(), ()>(&Arc::new(()), &tracker, RequestKind::Query, |_| {
                calls += 1;
                Box::pin(async {
                    Err(anyhow::Error::new(ConnectorError::TooManyRequestsError {
                        msg: "Too much request weight used.".to_string(),
                        code: Some(-1003),
                        retry_after: None,
                    }))
                })
            })
            .await;

        assert_eq!(calls, MAX_RATE_LIMIT_RETRIES + 1);
        let error = result.err().expect("every attempt was refused");
        assert!(
            matches!(
                error.downcast_ref::<ConnectorError>(),
                Some(ConnectorError::TooManyRequestsError { .. })
            ),
            "{error:?}"
        );
        // Each of the three sleeps wakes up to a millisecond late.
        let backoffs = Duration::from_secs(1 + 2 + 4);
        let elapsed = started.elapsed();
        assert!(
            elapsed >= backoffs && elapsed <= backoffs + Duration::from_millis(3),
            "{elapsed:?}"
        );
        assert_eq!(
            tracker.cooldown_remaining(),
            Some(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS))
        );
    }

    /// A request held back by a cooldown reports the time left as its wait.
    #[test]
    fn a_held_back_request_reports_the_cooldown_left() {
        let remaining = Duration::from_secs(42);
        let held = || anyhow::Error::new(CooldownActive { remaining });
        let order = classify_rest_order_error(&held(), &InstrumentNameExchange::new("BTCUSDT"));
        assert_eq!(order_retry_after(&order), Some(Some(remaining)));
        let query = classify_rest_query_error(&held(), None);
        assert!(
            matches!(
                &query,
                UnindexedClientError::Api(api @ ApiError::RateLimit { .. })
                    if api.retry_after() == Some(remaining)
            ),
            "{query:?}"
        );
    }

    /// A WS-API response replaces the default limit, so the same usage can cross the threshold
    /// under the reported limit that it stays under by default. A zero limit is ignored.
    #[tokio::test]
    async fn the_ws_api_limit_replaces_the_default() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.observe_ws_api(Some(&ws_weight(0, 1_000)), unix_ms());
        assert_eq!(
            tracker.weight_limit.load(Ordering::Relaxed),
            SPOT_REQUEST_WEIGHT_PER_MINUTE
        );
        assert!(!waits(&tracker, RequestKind::Query).await);

        tracker.observe_ws_api(Some(&ws_weight(1_100, 1_000)), unix_ms());
        assert_eq!(tracker.weight_limit.load(Ordering::Relaxed), 1_100);
        assert!(waits(&tracker, RequestKind::Query).await);
    }

    /// A late response, for a request sent in an earlier minute, pauses nothing.
    #[tokio::test]
    async fn a_response_from_an_earlier_minute_pauses_nothing() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.observe_ws_api(Some(&ws_weight(1_000, 1_000)), unix_ms() - 120_000);
        assert!(!waits(&tracker, RequestKind::Query).await);
    }

    /// A margin tracker ignores WS-API usage, which is spot weight. A debug build asserts that it
    /// is never given any.
    #[tokio::test]
    #[cfg_attr(
        debug_assertions,
        should_panic(expected = "WS-API usage is spot weight")
    )]
    async fn a_margin_tracker_ignores_ws_api_usage() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Sapi);
        tracker.observe_ws_api(Some(&ws_weight(1_000, 1_000)), unix_ms());
        assert_eq!(
            tracker.weight_limit.load(Ordering::Relaxed),
            SAPI_IP_WEIGHT_PER_MINUTE
        );
        assert!(!waits(&tracker, RequestKind::Query).await);
    }

    /// A response without the weight-per-minute entry changes nothing.
    #[tokio::test]
    async fn a_response_without_weight_changes_nothing() {
        tokio::time::pause();
        let tracker = RateLimitTracker::new(WeightPool::Spot);
        tracker.observe_ws_api(None, unix_ms());
        tracker.observe_ws_api(Some(&ws_weight(1_000, 1_000)[..1]), unix_ms());
        assert_eq!(
            tracker.weight_limit.load(Ordering::Relaxed),
            SPOT_REQUEST_WEIGHT_PER_MINUTE
        );
        assert!(!waits(&tracker, RequestKind::Query).await);
    }

    /// A handshake Binance refused with 429 is a throttle and with 418 a ban, which waits out
    /// Binance's shortest; other failures are neither.
    #[test]
    fn handshake_rate_limit_is_recognised() {
        let handshake = |msg: &str| anyhow::Error::new(WebsocketError::Handshake(msg.into()));
        let throttled = HandshakeRefusal::of(&handshake("HTTP error: 429 Too Many Requests"));
        assert_eq!(throttled, Some(HandshakeRefusal::Throttled));
        assert_eq!(HandshakeRefusal::Throttled.retry_after(), None);
        let banned = HandshakeRefusal::of(&handshake("HTTP error: 418 I'm a teapot"));
        assert_eq!(banned, Some(HandshakeRefusal::Banned));
        assert_eq!(
            HandshakeRefusal::Banned.retry_after(),
            Some(Duration::from_secs(120))
        );
        assert_eq!(
            HandshakeRefusal::of(
                &handshake("HTTP error: 429 Too Many Requests").context("connecting")
            ),
            Some(HandshakeRefusal::Throttled)
        );
        for other in [
            handshake("HTTP error: 503 Service Unavailable"),
            anyhow::Error::new(WebsocketError::Timeout),
            anyhow::anyhow!("HTTP error: 429 Too Many Requests"),
        ] {
            assert_eq!(HandshakeRefusal::of(&other), None, "{other:#}");
        }
    }

    /// What placing an order reports, by the response's status: an order that ended in the
    /// response is reported as it ended, not as open.
    #[test]
    fn a_placement_response_is_read_from_its_status() {
        use rust_decimal_macros::dec;

        let btc = InstrumentNameExchange::new("BTCUSDT");
        let time = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
        let placed = |status: Option<&str>, filled: Option<&str>, quote: Option<&str>| {
            placed_order_state(
                ExchangeId::BinanceSpot,
                &btc,
                dec!(2),
                PlacementResponse {
                    status,
                    order_id: OrderId::new("7"),
                    time_exchange: time,
                    executed_qty: filled,
                    cumulative_quote_qty: quote,
                },
            )
        };
        let expired = |filled| {
            OrderState::Inactive(InactiveOrderState::Expired(Expired::new(
                OrderId::new("7"),
                time,
                filled,
            )))
        };
        let open = |filled| {
            OrderState::active(Open::new(
                VenueOrderId::Assigned(OrderId::new("7")),
                time,
                filled,
            ))
        };

        assert_eq!(
            placed(Some("EXPIRED"), Some("0"), Some("0")),
            expired(Some(Decimal::ZERO)),
            "an IOC or FOK order that found no liquidity"
        );
        assert_eq!(
            placed(Some("EXPIRED"), Some("1"), Some("100")),
            expired(Some(dec!(1))),
            "an IOC order that partly filled and expired the rest"
        );
        assert_eq!(
            placed(Some("EXPIRED_IN_MATCH"), Some("0"), None),
            expired(Some(Decimal::ZERO)),
            "expired by self-trade prevention"
        );
        assert_eq!(
            placed(Some("FILLED"), Some("2"), Some("201")),
            OrderState::fully_filled(Filled::new(
                OrderId::new("7"),
                time,
                dec!(2),
                Some(dec!(100.5))
            ))
        );
        assert_eq!(
            placed(Some("CANCELED"), Some("1"), None),
            OrderState::Inactive(InactiveOrderState::Cancelled(Cancelled::new(
                OrderId::new("7"),
                time,
                Some(dec!(1))
            )))
        );
        assert!(matches!(
            placed(Some("REJECTED"), Some("0"), None),
            OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Rejected(
                ApiError::OrderRejected(_)
            )))
        ));
        for status in [
            Some("NEW"),
            Some("PARTIALLY_FILLED"),
            Some("PENDING_NEW"),
            None,
            Some("FROZEN"),
        ] {
            assert_eq!(
                placed(status, Some("1"), Some("100")),
                open(dec!(1)),
                "{status:?} reads as open from what filled"
            );
        }
        assert!(
            matches!(
                placed(None, Some("2"), None),
                OrderState::Inactive(InactiveOrderState::FullyFilled(_))
            ),
            "an ACK that somehow reports everything filled"
        );
    }

    /// A placement response that does not say what filled leaves an ended order's fill unknown,
    /// not zero.
    #[test]
    fn a_placement_response_without_a_fill_reports_it_unknown() {
        use rust_decimal_macros::dec;

        let btc = InstrumentNameExchange::new("BTCUSDT");
        let time = Utc.timestamp_millis_opt(1_700_000_000_000).unwrap();
        let placed = |status: Option<&str>, filled: Option<&str>| {
            placed_order_state(
                ExchangeId::BinanceSpot,
                &btc,
                dec!(2),
                PlacementResponse {
                    status,
                    order_id: OrderId::new("7"),
                    time_exchange: time,
                    executed_qty: filled,
                    cumulative_quote_qty: Some("201"),
                },
            )
        };
        let id = || OrderId::new("7");

        for filled in [None, Some("not a number")] {
            assert_eq!(
                placed(Some("EXPIRED"), filled),
                OrderState::Inactive(InactiveOrderState::Expired(Expired::new(id(), time, None))),
                "{filled:?}"
            );
            assert_eq!(
                placed(Some("CANCELED"), filled),
                OrderState::Inactive(InactiveOrderState::Cancelled(Cancelled::new(
                    id(),
                    time,
                    None
                ))),
                "{filled:?}"
            );
            assert_eq!(
                placed(Some("FILLED"), filled),
                OrderState::fully_filled(Filled::new(id(), time, dec!(2), Some(dec!(100.5)))),
                "a filled order filled its whole quantity, at the quote over it: {filled:?}"
            );
            assert_eq!(
                placed(Some("NEW"), filled),
                OrderState::active(Open::new(VenueOrderId::Assigned(id()), time, Decimal::ZERO)),
                "a live order's unknown fill reads as nothing filled yet: {filled:?}"
            );
        }
        assert_eq!(
            placed(None, None),
            OrderState::active(Open::new(VenueOrderId::Assigned(id()), time, Decimal::ZERO)),
            "an ACK reports neither status nor fill"
        );
    }

    #[test]
    fn the_average_price_is_the_quote_traded_over_the_base() {
        use rust_decimal_macros::dec;

        let id = OrderId::new("7");
        let avg = |quote, filled| binance_avg_price(ExchangeId::BinanceSpot, &id, quote, filled);
        assert_eq!(avg(Some("100"), dec!(4)), Some(dec!(25)));
        assert_eq!(avg(Some("100"), Decimal::ZERO), None, "nothing filled");
        assert_eq!(avg(None, dec!(4)), None, "no quote");
        assert_eq!(
            avg(Some("-1"), dec!(4)),
            None,
            "Binance's \"not available\""
        );
        assert_eq!(avg(Some("not-a-number"), dec!(4)), None);
    }
}

/// A local WS API for the account-stream managers' tests, and the reconnect-budget checks spot and
/// margin share.
#[cfg(test)]
#[allow(clippy::unwrap_used)] // Test code: panics on bad input are acceptable
pub(super) mod test_ws_api {
    use super::HEARTBEAT_TIMEOUT_SECS;
    use crate::client::reconnect_budget::BackoffLimits;
    use futures::{SinkExt as _, StreamExt as _};
    use parking_lot::Mutex;
    use std::{sync::Arc, time::Duration};
    use tokio::time::Instant;
    use tokio_tungstenite::tungstenite::Message;

    /// How the server treats a connection after its first answer.
    #[derive(Debug, Clone, Copy)]
    pub(in crate::client::binance) enum Lifetime {
        /// Stays open, answering every request.
        Open,
        /// Closes after this long, sending nothing meanwhile.
        Silent(Duration),
        /// Closes after this long, pinging every second meanwhile.
        Pinging(Duration),
    }

    /// Starts a WS API answering every request with `answer`, whose `n`th connection (from 0)
    /// lives `lifetime(n)` after its first answer. Returns its `ws://` URL, and when each request
    /// arrived.
    pub(in crate::client::binance) async fn serve(
        answer: fn(&serde_json::Value) -> serde_json::Value,
        lifetime: fn(usize) -> Lifetime,
    ) -> (String, Arc<Mutex<Vec<Instant>>>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("ws://{}", listener.local_addr().unwrap());
        let requests = Arc::new(Mutex::new(Vec::new()));
        let seen = Arc::clone(&requests);
        tokio::spawn(async move {
            let mut connections = 0..;
            while let Ok((stream, _)) = listener.accept().await {
                // Without it, a frame can wait in the kernel for the client's delayed ACK, which is
                // real time a paused clock does not wait for.
                stream.set_nodelay(true).unwrap();
                let seen = Arc::clone(&seen);
                let lifetime = connections.next().map_or(Lifetime::Open, lifetime);
                tokio::spawn(async move {
                    let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
                        return;
                    };
                    while let Some(Ok(message)) = ws.next().await {
                        let Message::Text(text) = message else {
                            continue;
                        };
                        seen.lock().push(Instant::now());
                        let request: serde_json::Value = serde_json::from_str(&text).unwrap();
                        if ws
                            .send(Message::text(answer(&request).to_string()))
                            .await
                            .is_err()
                        {
                            return;
                        }
                        let (after, pinging) = match lifetime {
                            Lifetime::Open => continue,
                            Lifetime::Silent(after) => (after, false),
                            Lifetime::Pinging(after) => (after, true),
                        };
                        let closes = tokio::time::sleep(after);
                        tokio::pin!(closes);
                        let mut pings = tokio::time::interval(Duration::from_secs(1));
                        loop {
                            tokio::select! {
                                () = &mut closes => break,
                                _ = pings.tick(), if pinging => {
                                    if ws.send(Message::Ping(Default::default())).await.is_err() {
                                        return;
                                    }
                                }
                            }
                        }
                        let _ = ws.close(None).await;
                        return;
                    }
                });
            }
        });
        (url, requests)
    }

    /// How long a stable connection lives: past the heartbeat tick that proves it.
    pub(in crate::client::binance) const STABLE: Duration =
        Duration::from_secs(HEARTBEAT_TIMEOUT_SECS + 3);

    /// How long a dropped connection lives.
    const DROPPED: Duration = Duration::from_millis(100);

    /// Two connections dropped at once, then one that pings past the heartbeat tick, then one that
    /// stays.
    pub(in crate::client::binance) fn dropped_then_pinging(n: usize) -> Lifetime {
        match n {
            0 | 1 => Lifetime::Silent(DROPPED),
            2 => Lifetime::Pinging(STABLE),
            _ => Lifetime::Open,
        }
    }

    /// A connection dropped at once, then one silent past the heartbeat tick, then one that stays.
    pub(in crate::client::binance) fn dropped_then_silent(n: usize) -> Lifetime {
        match n {
            0 => Lifetime::Silent(DROPPED),
            1 => Lifetime::Silent(STABLE),
            _ => Lifetime::Open,
        }
    }

    /// Waits for the `count`th request, and returns when each arrived.
    async fn requests_until(
        requests: &Mutex<Vec<Instant>>,
        count: usize,
        label: &str,
    ) -> Vec<Instant> {
        let deadline = Instant::now() + 2 * STABLE + Duration::from_secs(15);
        while requests.lock().len() < count {
            assert!(
                Instant::now() < deadline,
                "{label}: {} requests",
                requests.lock().len()
            );
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        requests.lock().clone()
    }

    /// Checks a manager serving [`dropped_then_pinging`], on a paused clock so each wait is exact.
    /// The subscribe on the second dropped connection does not clear the budget, so its drop waits
    /// the second backoff. The pinged connection's heartbeat tick does, so its drop waits the first
    /// again, not the third.
    pub(in crate::client::binance) async fn assert_reset_once_pinged(
        requests: &Mutex<Vec<Instant>>,
        label: &str,
    ) {
        let initial = BackoffLimits::ACCOUNT_STREAM.initial;
        assert_eq!(
            waits(&requests_until(requests, 4, label).await),
            [DROPPED + initial, DROPPED + 2 * initial, STABLE + initial],
            "{label}: the waits between connections"
        );
    }

    /// Checks a manager serving [`dropped_then_silent`], on a paused clock so each wait is exact. A
    /// connection that outlives the first heartbeat tick with nothing received does not clear the
    /// budget, so its drop waits the second backoff.
    pub(in crate::client::binance) async fn assert_no_reset_when_silent(
        requests: &Mutex<Vec<Instant>>,
        label: &str,
    ) {
        let initial = BackoffLimits::ACCOUNT_STREAM.initial;
        assert_eq!(
            waits(&requests_until(requests, 3, label).await),
            [DROPPED + initial, STABLE + 2 * initial],
            "{label}: the waits between connections"
        );
    }

    /// The time between each request and the next.
    fn waits(at: &[Instant]) -> Vec<Duration> {
        at.windows(2).map(|w| w[1] - w[0]).collect()
    }
}