rustrade-execution 0.15.0

//! Shared utilities for Hyperliquid execution clients (perps and spot).
//!
//! Contains parsing helpers and stream wrappers used by both
//! `HyperliquidClient` (perps) and `HyperliquidSpotClient`.
//!
//! Error mapping is in the `error` module.

use crate::client::hyperliquid::error::{map_sdk_error, order_rejection};
use crate::error::{OrderError, UnindexedClientError};
use crate::order::{
    Order, OrderKey, OrderKind, TimeInForce, UnindexedOrderSnapshot,
    id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
    state::{Cancelled, Filled, Open, OrderState},
};
use crate::{
    InstrumentAccountSnapshot, UnindexedAccountEvent,
    balance::{AssetBalance, Balance},
    position::PositionReport,
};
use chrono::{DateTime, TimeZone, Utc};
use fnv::FnvHashSet;
use futures::Stream;
use hyperliquid_rust_sdk::{ExchangeDataStatus, ExchangeResponseStatus};
use rust_decimal::Decimal;
use rustrade_instrument::{
    Side,
    asset::name::AssetNameExchange,
    exchange::ExchangeId,
    hyperliquid::{CoinKind, SpotPair},
    instrument::name::InstrumentNameExchange,
};
use serde::de::DeserializeOwned;
use smol_str::format_smolstr;
use std::collections::{BTreeSet, HashMap, HashSet};
use std::pin::Pin;
use std::str::FromStr;
use std::task::{Context, Poll};
use tokio_util::sync::CancellationToken;
use tracing::warn;
use uuid::Uuid;

/// Stream wrapper that cancels background tasks when dropped.
///
/// Ensures spawned WebSocket processing tasks are cleaned up when the consumer
/// drops the account stream, preventing task leaks.
#[derive(Debug)]
pub struct CancelOnDropStream<S> {
    inner: S,
    cancel_token: CancellationToken,
}

impl<S> CancelOnDropStream<S> {
    /// Create a new cancel-on-drop stream wrapper.
    pub(crate) fn new(inner: S, cancel_token: CancellationToken) -> Self {
        Self {
            inner,
            cancel_token,
        }
    }
}

impl<S: Stream + Unpin> Stream for CancelOnDropStream<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)
    }
}

impl<S> Drop for CancelOnDropStream<S> {
    fn drop(&mut self) {
        self.cancel_token.cancel();
    }
}

/// Parse a decimal string from SDK response, logging warnings on failure.
/// One fill from the `userFills` info endpoint, as the venue actually returns it.
///
/// # Why this exists rather than `hyperliquid_rust_sdk::UserFillsResponse`
///
/// The SDK's type does not model `tid`, the only per-fill identifier the endpoint offers, and it
/// does not model `feeToken`. Neither omission is visible at runtime: the SDK does not set
/// `deny_unknown_fields`, so both are parsed away silently. Without `tid` the only id left is
/// `hash`, which identifies the *transaction* — one aggressive order sweeping several resting
/// orders produces several fills under a single hash, and they are then indistinguishable.
///
/// Only the fields this client reads are declared. Unknown fields are ignored, so `hash`,
/// `closedPnl`, `dir`, `startPosition`, `crossed` and `builderFee` cost nothing by being absent
/// here.
///
/// Drop this in favour of the SDK type if it ever models `tid`.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct UserFill {
    pub coin: String,
    pub side: String,
    pub px: String,
    pub sz: String,
    pub time: u64,
    pub oid: u64,
    pub fee: String,

    /// Identifies this fill. `hash` identifies the transaction that contained it, and one
    /// transaction can contain many fills.
    ///
    /// Declared required rather than `Option<u64>` deliberately. If the venue ever stops sending
    /// it, this client should fail loudly at the parse — falling back to `hash` would silently
    /// restore a duplicate-id defect that is invisible until fills go missing during
    /// reconciliation.
    ///
    /// Hyperliquid documents this as unique per fill, but qualified by coin rather than global,
    /// which is why the dedup cache keys on the instrument alongside it.
    pub tid: u64,

    /// Asset the fee is denominated in.
    ///
    /// Optional because, unlike every other field here, nothing proves the endpoint sends it: the
    /// SDK's own type omits it, so it has never been observed through this dependency. The caller
    /// falls back to inferring the asset from the side when it is absent.
    pub fee_token: Option<String>,
    /// The cloid of the order filled, in the venue's form ([`cloid_to_cid`]). Absent for an order
    /// placed without one.
    pub cloid: Option<String>,
}

/// The span `start..=end` in epoch milliseconds, as `userFillsByTime` takes it, or `None` when it
/// is empty or ends before the epoch. Fill times are whole milliseconds, read from the
/// millisecond holding `start`; the caller applies the exact span to the fills it reads.
pub(super) fn span_millis(start: DateTime<Utc>, end: DateTime<Utc>) -> Option<(u64, u64)> {
    if start > end {
        return None;
    }
    let end_ms = u64::try_from(end.timestamp_millis()).ok()?;
    let start_ms = u64::try_from(start.timestamp_millis()).unwrap_or(0);
    Some((start_ms, end_ms))
}

/// The most fills one `userFillsByTime` response holds.
const USER_FILLS_PER_RESPONSE: usize = 2_000;

/// How many of a wallet's most recent fills Hyperliquid keeps.
const FILLS_RETAINED: usize = 10_000;

/// When the 10,000-fill limit cuts into a span that `fills` (everything `userFillsByTime` returned
/// for it, before any filter by instrument) was read from `start_ms`: the time of the oldest fill
/// held, when a full 10,000 came back and it is after `start_ms`. Fewer than 10,000 means nothing
/// has been dropped, however old the span.
pub(super) fn held_since(fills: &[UserFill], start_ms: u64) -> Option<DateTime<Utc>> {
    if fills.len() < FILLS_RETAINED {
        return None;
    }
    let oldest = fills.iter().map(|fill| fill.time).min()?;
    (oldest > start_ms).then(|| {
        Utc.timestamp_millis_opt(i64::try_from(oldest).ok()?)
            .single()
    })?
}

/// Read every fill `address` made from `start_ms` to `end_ms`, both inclusive, in epoch
/// milliseconds, parsed into [`UserFill`] rather than the SDK's lossy type.
///
/// `userFillsByTime` returns at most 2,000 fills per response, oldest
/// first, with both bounds inclusive (observed on mainnet, 2026-10-05). A full response is
/// followed by a read from its last fill's time, which returns again the fills at that
/// millisecond already read; those are dropped by coin and `tid`. Hyperliquid keeps only each
/// wallet's 10,000 most recent fills, so the whole read is a handful of requests, and it cannot
/// reach a fill older than those.
///
/// `userFills`, which takes no span, is not used: it returns only the 2,000 most recent fills, so
/// a span holding more would come back short without saying so.
///
/// # Errors
///
/// Returns a transport error if a POST fails, or a parse error if a response does not match
/// [`UserFill`] — which includes the case where `tid` has stopped being sent.
/// [`Truncated`](UnindexedClientError::Truncated) when a full response holds only fills at the
/// millisecond it was read from, so the read cannot advance.
pub async fn user_fills_by_time(
    info_client: &hyperliquid_rust_sdk::InfoClient,
    address: ethers::types::H160,
    start_ms: u64,
    end_ms: u64,
) -> Result<Vec<UserFill>, UnindexedClientError> {
    let mut fills: Vec<UserFill> = Vec::new();
    let mut from = start_ms;
    loop {
        // `{:?}` on H160 renders the full checksummed form; see `user_info`.
        let body = format!(
            r#"{{"type":"userFillsByTime","user":"{address:?}","startTime":{from},"endTime":{end_ms}}}"#
        );
        let page: Vec<UserFill> = info(info_client, "userFillsByTime", body).await?;
        let full = page.len() >= USER_FILLS_PER_RESPONSE;
        let Some(last_time) = page.last().map(|fill| fill.time) else {
            break;
        };
        // The fills at `from` that the previous response already returned: a few at most, since
        // they share one millisecond.
        let seen: Vec<(String, u64)> = fills
            .iter()
            .rev()
            .take_while(|fill| fill.time == from)
            .map(|fill| (fill.coin.clone(), fill.tid))
            .collect();
        fills.extend(page.into_iter().filter(|fill| {
            fill.time != from
                || !seen
                    .iter()
                    .any(|(coin, tid)| *tid == fill.tid && *coin == fill.coin)
        }));
        if !full {
            break;
        }
        if last_time <= from {
            return Err(UnindexedClientError::Truncated {
                fills_read: fills.len(),
            });
        }
        from = last_time;
    }
    Ok(fills)
}

/// One order from the `openOrders` info endpoint, as the venue actually returns it.
///
/// # Why this exists rather than `hyperliquid_rust_sdk::OpenOrdersResponse`
///
/// The SDK's type does not model `cloid`, the client order id the order was placed with, nor
/// `origSz`. The venue sends both, although its documentation lists neither. Without `cloid` an
/// order can only be reported under its venue `oid`, which is not the id the engine tracks it by.
///
/// Drop this in favour of the SDK type if it ever models `cloid`.
#[derive(Debug, Clone, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct OpenOrder {
    pub coin: String,
    pub side: String,
    pub limit_px: String,
    /// Quantity still to fill.
    pub sz: String,
    pub oid: u64,
    pub timestamp: u64,

    /// Quantity the order was placed for.
    ///
    /// Optional because the endpoint's documentation does not list it. Without it the order's
    /// filled quantity is unknown, and it is reported as its remaining quantity with nothing
    /// filled.
    #[serde(default)]
    pub orig_sz: Option<String>,

    /// The client order id the order was placed with, as `0x` and 32 hex digits (see
    /// [`cloid_to_cid`]). Absent for an order placed without one.
    #[serde(default)]
    pub cloid: Option<String>,
}

/// Fetch every open order for `address`, parsed into [`OpenOrder`] rather than the SDK's lossy
/// type, which drops the `cloid`.
///
/// # Errors
///
/// Returns a transport error if the POST fails, or a parse error if the response does not match
/// [`OpenOrder`].
pub async fn open_orders(
    info_client: &hyperliquid_rust_sdk::InfoClient,
    address: ethers::types::H160,
) -> Result<Vec<OpenOrder>, UnindexedClientError> {
    user_info(info_client, "openOrders", address, None).await
}

/// POST the info request `{"type": kind, "user": address}`, with `"dex": dex` when `dex` names a
/// builder-deployed (HIP-3) perpetual DEX, and parse the response as `T`.
///
/// Issued through the SDK's own [`HttpClient`](hyperliquid_rust_sdk::InfoClient::http_client), so
/// base URL, TLS configuration and error type are exactly those of every other info request this
/// client makes — only the response type differs.
pub(super) async fn user_info<T: DeserializeOwned>(
    info_client: &hyperliquid_rust_sdk::InfoClient,
    kind: &str,
    address: ethers::types::H160,
    dex: Option<&str>,
) -> Result<T, UnindexedClientError> {
    // `{:?}` on H160 renders the checksummed 0x-prefixed form the endpoint expects. `Display`
    // abbreviates the middle of the address ("0x1234…5678"), so it must not be used here.
    let mut body = serde_json::json!({"type": kind, "user": format!("{address:?}")});
    if let Some(dex) = dex {
        body["dex"] = dex.into();
    }
    info(info_client, kind, body.to_string()).await
}

/// POST the info request `body`, of type `kind`, and parse the response as `T`.
pub(super) async fn info<T: DeserializeOwned>(
    info_client: &hyperliquid_rust_sdk::InfoClient,
    kind: &str,
    body: String,
) -> Result<T, UnindexedClientError> {
    let raw = info_client
        .http_client
        .post("/info", body)
        .await
        .map_err(map_sdk_error)?;

    // `Internal` rather than a connectivity error: a response that does not parse is a schema
    // change or a venue-side regression, not a transient fault, and retrying will not fix it.
    serde_json::from_str(&raw).map_err(|e| {
        UnindexedClientError::Internal(format!("Hyperliquid {kind} response did not parse: {e}"))
    })
}

/// Each spot token balance in `balances` (`spotClearinghouseState`), free of what is on hold.
pub(super) fn spot_balances<'a>(
    balances: impl IntoIterator<Item = &'a hyperliquid_rust_sdk::UserTokenBalance>,
    now: DateTime<Utc>,
) -> impl Iterator<Item = AssetBalance<AssetNameExchange>> {
    balances.into_iter().map(move |balance| {
        let total = parse_decimal(&balance.total, "total").unwrap_or(Decimal::ZERO);
        let hold = parse_decimal(&balance.hold, "hold").unwrap_or(Decimal::ZERO);
        let free = (total - hold).max(Decimal::ZERO);
        AssetBalance::new(
            AssetNameExchange::from(balance.coin.as_str()),
            Balance::new(total, free),
            now,
        )
    })
}

/// Convert one [`OpenOrder`] into an `Open` order on `instrument`, keyed as [`record_cid`] says.
///
/// `None`, with a `warn!`, when a field does not parse. The caller then cannot say the order is
/// not open, so it must not declare its list complete.
pub(super) fn open_order_to_order(
    row: &OpenOrder,
    exchange: ExchangeId,
    instrument: InstrumentNameExchange,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
    let parsed = (|| {
        let cid = record_cid(row.cloid.as_deref(), row.oid)?;
        let side = parse_side(&row.side)?;
        let price = parse_decimal(&row.limit_px, "limitPx")?;
        let remaining = parse_decimal(&row.sz, "sz")?;
        let quantity = match &row.orig_sz {
            Some(orig_sz) => parse_decimal(orig_sz, "origSz")?,
            None => remaining,
        };
        let time_exchange = millis_to_datetime(row.timestamp)?;
        Some((cid, side, price, quantity, remaining, time_exchange))
    })();

    let Some((cid, side, price, quantity, remaining, time_exchange)) = parsed else {
        warn!(
            %exchange,
            %instrument,
            oid = row.oid,
            order = ?row,
            "Hyperliquid open order did not convert - leaving it out, so this instrument's order \
             list is not complete"
        );
        return None;
    };

    Some(Order {
        key: OrderKey {
            exchange,
            instrument,
            strategy: StrategyId::unknown(),
            cid,
        },
        side,
        price: Some(price),
        quantity,
        // Neither the kind nor the time in force is in this response.
        kind: OrderKind::Limit,
        time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
        state: Open {
            id: VenueOrderId::Assigned(OrderId(format_smolstr!("{}", row.oid))),
            time_exchange,
            filled_quantity: (quantity - remaining).max(Decimal::ZERO),
        },
    })
}

/// Which instruments a read keeps: those in the slice it was given, or every one when the slice is
/// empty, the "return all" sentinel of [`ExecutionClient`](crate::client::ExecutionClient)'s reads.
pub(super) struct InstrumentFilter<'a>(Option<FnvHashSet<&'a InstrumentNameExchange>>);

impl<'a> InstrumentFilter<'a> {
    pub(super) fn new(instruments: &'a [InstrumentNameExchange]) -> Self {
        Self((!instruments.is_empty()).then(|| instruments.iter().collect()))
    }

    /// Whether the read keeps `instrument`.
    pub(super) fn matches(&self, instrument: &InstrumentNameExchange) -> bool {
        self.0.as_ref().is_none_or(|set| set.contains(instrument))
    }
}

type UnindexedInstrumentAccountSnapshot =
    InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange>;

/// The open orders of an account snapshot, one entry per instrument.
pub(super) struct OpenOrderListing(
    HashMap<InstrumentNameExchange, UnindexedInstrumentAccountSnapshot>,
);

impl OpenOrderListing {
    /// List `rows` by instrument, keeping only the rows `to_instrument` maps to an instrument in
    /// `requested` (every instrument when `requested` is empty).
    ///
    /// Each entry declares its orders complete unless one of its rows failed to convert. That
    /// holds because the endpoint returns every open order, unpaged, and because [`record_cid`]
    /// reports each order placed through this client under the id it was placed with.
    ///
    /// Every requested instrument gets an entry, open orders or not. An entry is what lets the
    /// engine judge an order that is no longer listed to be gone, and the instrument whose last
    /// order was cancelled while the account stream was down is exactly the one with nothing
    /// left to list. With `requested` empty there is nothing to enumerate, so only instruments
    /// with open orders get one.
    pub(super) fn new(
        rows: &[OpenOrder],
        exchange: ExchangeId,
        requested: &[InstrumentNameExchange],
        to_instrument: impl Fn(&str) -> Option<InstrumentNameExchange>,
    ) -> Self {
        fn entry<'a>(
            listing: &'a mut HashMap<InstrumentNameExchange, UnindexedInstrumentAccountSnapshot>,
            instrument: &InstrumentNameExchange,
        ) -> &'a mut UnindexedInstrumentAccountSnapshot {
            listing
                .entry(instrument.clone())
                .or_insert_with(|| InstrumentAccountSnapshot {
                    instrument: instrument.clone(),
                    orders: Vec::new(),
                    orders_complete: true,
                    position: PositionReport::Unreported,
                    isolated: None,
                })
        }

        let filter = InstrumentFilter::new(requested);
        let mut listing = HashMap::with_capacity(requested.len());

        for instrument in requested {
            entry(&mut listing, instrument);
        }

        for row in rows {
            let Some(instrument) = to_instrument(&row.coin) else {
                continue;
            };
            if !filter.matches(&instrument) {
                continue;
            }
            let order = open_order_to_order(row, exchange, instrument.clone());
            let snapshot = entry(&mut listing, &instrument);
            match order {
                Some(order) => snapshot.orders.push(order.into()),
                None => snapshot.orders_complete = false,
            }
        }

        Self(listing)
    }

    /// Take the entry for `instrument`, if it has one.
    pub(super) fn remove(
        &mut self,
        instrument: &InstrumentNameExchange,
    ) -> Option<UnindexedInstrumentAccountSnapshot> {
        self.0.remove(instrument)
    }

    /// Every remaining entry.
    pub(super) fn into_snapshots(self) -> impl Iterator<Item = UnindexedInstrumentAccountSnapshot> {
        self.0.into_values()
    }
}

/// What a Hyperliquid order status says about the order.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OrderStatus {
    Open,
    Filled,
    Cancelled,
    Rejected,
}

impl OrderStatus {
    /// Classify an `orderUpdates` / `orderStatus` status string.
    ///
    /// Beyond `open`, `filled`, `canceled` and `rejected`, Hyperliquid names the reason for each
    /// other way an order ends: `marginCanceled`, `selfTradeCanceled`, `reduceOnlyCanceled`,
    /// `tickRejected`, `badAloPxRejected` and a dozen more. They are classified by their suffix
    /// so that a reason added later still ends the order, rather than leaving it tracked as open
    /// for good. `scheduledCancel` is the one ending that breaks the pattern.
    ///
    /// `triggered` is a trigger order whose condition was met. It is still working, and ends
    /// with a later status of its own.
    fn classify(status: &str) -> Option<Self> {
        match status {
            "open" | "triggered" => Some(Self::Open),
            "filled" => Some(Self::Filled),
            "canceled" | "scheduledCancel" => Some(Self::Cancelled),
            "rejected" => Some(Self::Rejected),
            status if status.ends_with("Canceled") => Some(Self::Cancelled),
            status if status.ends_with("Rejected") => Some(Self::Rejected),
            _ => None,
        }
    }
}

/// Convert an `orderUpdates` message into an order snapshot event on `instrument`, keyed as
/// [`record_cid`] says.
///
/// `None`, with a `warn!` naming the cause where the parse helpers do not, when a field does not
/// parse or the status is not one [`OrderStatus::classify`] recognises.
pub(super) fn order_update_to_account_event(
    update: &hyperliquid_rust_sdk::OrderUpdate,
    exchange: ExchangeId,
    instrument: InstrumentNameExchange,
) -> Option<UnindexedAccountEvent> {
    let snapshot = order_update_to_order(update, exchange, instrument)?;
    Some(crate::AccountEvent::new(
        exchange,
        crate::AccountEventKind::OrderSnapshot(
            rustrade_integration::collection::snapshot::Snapshot(snapshot),
        ),
    ))
}

/// Convert an order record, as `orderUpdates` sends it and `orderStatus` answers with it, into the
/// order it describes on `instrument`, keyed as [`record_cid`] says.
///
/// `None`, with a `warn!` naming the cause where the parse helpers do not, when a field does not
/// parse or the status is not one [`OrderStatus::classify`] recognises.
pub(super) fn order_update_to_order(
    update: &hyperliquid_rust_sdk::OrderUpdate,
    exchange: ExchangeId,
    instrument: InstrumentNameExchange,
) -> Option<UnindexedOrderSnapshot> {
    let order = &update.order;
    let Some(status) = OrderStatus::classify(&update.status) else {
        warn!(%exchange, status = %update.status, oid = order.oid, "Unknown Hyperliquid order status - ignoring the update");
        return None;
    };
    let cid = record_cid(order.cloid.as_deref(), order.oid)?;
    let side = parse_side(&order.side)?;
    let price = parse_decimal(&order.limit_px, "order.limitPx")?;
    let orig_sz = parse_decimal(&order.orig_sz, "order.origSz")?;
    let Some(time_exchange) = millis_to_datetime(update.status_timestamp) else {
        warn!(%exchange, oid = order.oid, status_timestamp = update.status_timestamp, "Invalid Hyperliquid order update timestamp - ignoring the update");
        return None;
    };
    let order_id = OrderId(format_smolstr!("{}", order.oid));
    let filled_quantity = || {
        parse_decimal(&order.sz, "order.sz")
            .map(|remaining| (orig_sz - remaining).max(Decimal::ZERO))
    };

    let state = match status {
        OrderStatus::Open => OrderState::active(Open {
            id: VenueOrderId::Assigned(order_id),
            time_exchange,
            filled_quantity: filled_quantity()?,
        }),
        OrderStatus::Filled => OrderState::fully_filled(Filled::new(
            order_id,
            time_exchange,
            orig_sz,
            None, // The update does not carry the average price.
        )),
        // An unparseable remaining size leaves the fill unknown, not the cancel.
        OrderStatus::Cancelled => {
            OrderState::inactive(Cancelled::new(order_id, time_exchange, filled_quantity()))
        }
        OrderStatus::Rejected => {
            OrderState::inactive(OrderError::Rejected(order_rejection(update.status.clone())))
        }
    };

    // The update carries neither the order's kind nor its time in force.
    Some(Order {
        key: OrderKey {
            exchange,
            instrument,
            strategy: StrategyId::unknown(),
            cid,
        },
        side,
        price: Some(price),
        quantity: orig_sz,
        kind: OrderKind::Limit,
        time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
        state,
    })
}

pub fn parse_decimal(value: &str, field: &str) -> Option<Decimal> {
    Decimal::from_str(value)
        .map_err(|e| warn!(%field, %value, %e, "Failed to parse decimal"))
        .ok()
}

/// Parse SDK side string to rustrade Side.
///
/// Hyperliquid API returns "B" for buy/long and "A" for sell/short (ask-side).
/// Extra variants included for defensive parsing of potential future API changes.
pub fn parse_side(side: &str) -> Option<Side> {
    match side {
        "B" | "b" | "BUY" | "Buy" | "buy" => Some(Side::Buy),
        "A" | "a" | "S" | "s" | "SELL" | "Sell" | "sell" => Some(Side::Sell),
        _ => {
            warn!(%side, "Unknown side string");
            None
        }
    }
}

/// Convert milliseconds timestamp to `DateTime<Utc>`.
///
/// Returns `None` for timestamps outside the representable range (year 292M+).
pub fn millis_to_datetime(millis: u64) -> Option<DateTime<Utc>> {
    Utc.timestamp_millis_opt(i64::try_from(millis).ok()?)
        .single()
}

/// Map rustrade TimeInForce to Hyperliquid TIF string.
pub fn map_tif(tif: &TimeInForce) -> &'static str {
    match tif {
        TimeInForce::GoodUntilCancelled { post_only: true } => "Alo",
        TimeInForce::GoodUntilCancelled { post_only: false } => "Gtc",
        TimeInForce::ImmediateOrCancel => "Ioc",
        TimeInForce::FillOrKill => "Ioc", // Hyperliquid doesn't have FOK, use IOC
        TimeInForce::GoodUntilEndOfDay => "Gtc", // No EOD on Hyperliquid
        // Hyperliquid is a perpetuals DEX with no auction sessions and no native
        // GTD. Coerce to GTC and surface the loss of semantics so downstream
        // wrappers can decide whether to reject upstream.
        TimeInForce::GoodTillDate { .. } | TimeInForce::AtOpen | TimeInForce::AtClose => {
            warn!(time_in_force = ?tif, "Hyperliquid does not support this TimeInForce; coercing to Gtc");
            "Gtc"
        }
    }
}

/// Why an order is refused when its client id cannot be a cloid ([`cid_to_cloid`]).
pub(super) const CLOID_REQUIRED: &str = "Hyperliquid requires the client order id to be a UUID in \
     canonical form, as ClientOrderId::uuid() makes: the venue reports each order under that id, \
     and an order under any other id could not be matched to its own updates";

/// The cloid to place an order under `cid` with, or `None` if `cid` cannot be one.
///
/// Hyperliquid stores a cloid as 16 bytes and reports it back as `0x` and 32 lowercase hex
/// digits, which [`cloid_to_cid`] turns back into a client id. Only a `cid` that is the canonical
/// text of a UUID — lowercase and hyphenated, as [`ClientOrderId::uuid`] makes — comes back
/// unchanged. Any other spelling of a UUID would come back as a different id, under which the
/// order's own updates could not find it, so it is refused along with every id that is not a UUID.
pub fn cid_to_cloid(cid: &ClientOrderId) -> Option<Uuid> {
    let uuid = Uuid::try_parse(cid.0.as_str()).ok()?;
    let mut canonical = Uuid::encode_buffer();
    (uuid.hyphenated().encode_lower(&mut canonical) == cid.0.as_str()).then_some(uuid)
}

/// The client id an order placed with `cloid` was placed under: the inverse of [`cid_to_cloid`].
///
/// `cloid` is the venue's form, `0x` and 32 hex digits. `None` for anything else.
pub fn cloid_to_cid(cloid: &str) -> Option<ClientOrderId> {
    let hex = cloid.strip_prefix("0x")?;
    if hex.len() != 32 {
        return None;
    }
    // The 32-digit length check rules out the other spellings `try_parse` accepts.
    let uuid = Uuid::try_parse(hex).ok()?;
    Some(ClientOrderId::new(format_smolstr!("{}", uuid.hyphenated())))
}

/// Whether Hyperliquid applied a cancel of one order, read from its answer: `Err` with the reason
/// when it did not.
///
/// Hyperliquid answers a cancel it could not apply with a top-level `ok`, and puts the error in
/// the order's own status ("Order was never placed, already canceled, or filled.", for one), so
/// the top level alone does not say. Only a `success` status is a cancel. An answer carrying no
/// status, or another kind of status, is not taken as one either: the order may still be working.
/// A cancel names one order, so only the first status is read.
///
/// The reason is not classified. Hyperliquid's message says the order was never placed, already
/// cancelled, or filled, without saying which, so naming it already cancelled or already filled
/// would be a guess. A caller learns how the order ended from the account stream or a lookup.
pub(super) fn cancel_outcome(response: ExchangeResponseStatus) -> Result<(), String> {
    let response = match response {
        ExchangeResponseStatus::Ok(response) => response,
        ExchangeResponseStatus::Err(reason) => return Err(reason),
    };
    match response
        .data
        .and_then(|data| data.statuses.into_iter().next())
    {
        Some(ExchangeDataStatus::Success) => Ok(()),
        Some(ExchangeDataStatus::Error(reason)) => Err(reason),
        Some(status) => Err(format!("cancel answered with status {status:?}")),
        None => Err("cancel answered without a status for the order".to_owned()),
    }
}

/// The client id to report a venue order record under, given the record's `cloid` and `oid`.
///
/// Every order this client places carries a cloid ([`cid_to_cloid`]), so a record with one is
/// reported under the id the order was placed with. A record without one was placed some other
/// way, such as the web app, and has no client id at all. It is reported under its venue `oid`,
/// the only identifier it has, which is stable across reads so the order is tracked once.
///
/// `None`, with a `warn!`, for a cloid that is not in the venue's form. The record's order is
/// then unidentified, and must be left out rather than reported under a guess.
pub(super) fn record_cid(cloid: Option<&str>, oid: u64) -> Option<ClientOrderId> {
    match cloid {
        None => Some(ClientOrderId::new(format_smolstr!("{oid}"))),
        Some(cloid) => {
            let cid = cloid_to_cid(cloid);
            if cid.is_none() {
                warn!(%cloid, oid, "Hyperliquid order carries a cloid that is not 0x and 32 hex digits");
            }
            cid
        }
    }
}

/// Whether `coin` names a market of a kind [`CoinKind::of`] does not recognise.
fn is_unknown_coin(coin: &str) -> bool {
    !matches!(CoinKind::of(coin), CoinKind::Perp | CoinKind::Spot)
}

/// Log with one `warn!` the distinct coins among `coins` of a kind [`CoinKind::of`] does not
/// recognise, which neither client reports. For a read that lists many rows at once.
pub(super) fn warn_unknown_coins<'a>(coins: impl IntoIterator<Item = &'a str>) {
    let unknown: BTreeSet<&str> = coins
        .into_iter()
        .filter(|coin| is_unknown_coin(coin))
        .collect();
    if !unknown.is_empty() {
        warn!(
            ?unknown,
            "Hyperliquid coins name markets of an unrecognised kind; leaving them out"
        );
    }
}

/// The coins of a kind [`CoinKind::of`] does not recognise that an account stream has logged, so
/// it logs each once rather than on every event.
#[derive(Debug, Default)]
pub(super) struct UnknownCoins(HashSet<String>);

impl UnknownCoins {
    /// Log `coin` with `warn!` if it is of an unrecognised kind and not logged before.
    pub(super) fn warn_once(&mut self, coin: &str) {
        if is_unknown_coin(coin) && !self.0.contains(coin) {
            warn!(
                %coin,
                "Hyperliquid coin names a market of an unrecognised kind; leaving it out \
                 (logged once per stream)"
            );
            self.0.insert(coin.to_owned());
        }
    }
}

/// Build the spot instrument name for `pair`, from its tokens: `BASE-QUOTE-SPOT` (e.g.
/// `HYPE-USDC-SPOT` for the pair Hyperliquid names `@107`).
pub fn spot_pair_to_instrument(pair: &SpotPair) -> InstrumentNameExchange {
    InstrumentNameExchange::from(format_smolstr!("{}-{}-SPOT", pair.base(), pair.quote()))
}

/// Extract coin pair from spot instrument (e.g., "PURR-USDC-SPOT" -> "PURR/USDC").
///
/// Returns `Option<String>` — `None` if the instrument doesn't match expected
/// `BASE-QUOTE-SPOT` format. Callers should fail fast on `None` rather than
/// send a malformed asset to the exchange.
///
/// The SDK's `ExchangeClient` addresses every spot pair by this `BASE/QUOTE` form as well as by
/// the coin Hyperliquid names it (`@107`), but only spelled exactly as `spotMeta` spells the
/// pair's tokens. The name is not checked against `spotMeta`, so `KPEPE-USDC-SPOT` gives
/// `KPEPE/USDC`, which the SDK does not know for the pair `kPEPE/USDC`.
pub fn instrument_to_spot_coin(instrument: &InstrumentNameExchange) -> Option<String> {
    let (base, quote) = spot_base_quote(instrument)?;
    Some(format!("{base}/{quote}"))
}

/// The base and quote tokens of a spot instrument named `BASE-QUOTE-SPOT`, `None` for a name of
/// another form.
pub(super) fn spot_base_quote(instrument: &InstrumentNameExchange) -> Option<(&str, &str)> {
    instrument.as_ref().strip_suffix("-SPOT")?.split_once('-')
}

#[cfg(test)]
// Test code: panics on bad input are acceptable
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use rust_decimal_macros::dec;

    fn fills_at(times: impl IntoIterator<Item = u64>) -> Vec<UserFill> {
        times
            .into_iter()
            .map(|time| {
                serde_json::from_value(serde_json::json!({
                    "coin": "BTC", "side": "B", "px": "1", "sz": "1", "time": time,
                    "oid": 1, "fee": "0", "tid": time
                }))
                .unwrap()
            })
            .collect()
    }

    #[test]
    fn held_since_is_the_oldest_fill_only_when_the_retention_limit_is_reached() {
        let full = fills_at((0..FILLS_RETAINED as u64).map(|i| 5_000 + i));
        assert_eq!(
            held_since(&full, 1_000),
            Some(Utc.timestamp_millis_opt(5_000).unwrap())
        );

        let short = fills_at((0..FILLS_RETAINED as u64 - 1).map(|i| 5_000 + i));
        assert_eq!(held_since(&short, 1_000), None);
    }

    #[test]
    fn held_since_is_none_when_the_oldest_fill_is_not_after_the_span_start() {
        let full = fills_at((0..FILLS_RETAINED as u64).map(|i| 5_000 + i));

        assert_eq!(held_since(&full, 5_000), None);
        assert_eq!(held_since(&full, 9_000), None);
    }

    #[test]
    fn an_instrument_filter_keeps_what_it_was_given_or_everything_when_given_nothing() {
        let (btc, eth) = (
            InstrumentNameExchange::new("BTC"),
            InstrumentNameExchange::new("ETH"),
        );

        let all = InstrumentFilter::new(&[]);
        assert!(all.matches(&btc) && all.matches(&eth));

        let given = [btc.clone()];
        let one = InstrumentFilter::new(&given);
        assert!(one.matches(&btc));
        assert!(!one.matches(&eth));
    }

    #[test]
    fn test_parse_decimal_valid() {
        assert_eq!(parse_decimal("123.456", "test"), Some(dec!(123.456)));
        assert_eq!(parse_decimal("0", "test"), Some(dec!(0)));
        assert_eq!(parse_decimal("-50.5", "test"), Some(dec!(-50.5)));
    }

    #[test]
    fn test_parse_decimal_invalid() {
        assert_eq!(parse_decimal("", "test"), None);
        assert_eq!(parse_decimal("abc", "test"), None);
        assert_eq!(parse_decimal("12.34.56", "test"), None);
    }

    #[test]
    fn test_parse_side() {
        assert_eq!(parse_side("B"), Some(Side::Buy));
        assert_eq!(parse_side("BUY"), Some(Side::Buy));
        assert_eq!(parse_side("buy"), Some(Side::Buy));
        assert_eq!(parse_side("A"), Some(Side::Sell));
        assert_eq!(parse_side("S"), Some(Side::Sell));
        assert_eq!(parse_side("SELL"), Some(Side::Sell));
        assert_eq!(parse_side("sell"), Some(Side::Sell));
        assert_eq!(parse_side("X"), None);
        assert_eq!(parse_side(""), None);
    }

    #[test]
    fn test_spot_pair_to_instrument() {
        let pairs = super::super::spot_coins::test_spot_pairs();
        let instrument = |coin| spot_pair_to_instrument(pairs.get(coin).unwrap());

        assert_eq!(instrument("PURR/USDC").as_ref(), "PURR-USDC-SPOT");
        assert_eq!(instrument("@107").as_ref(), "HYPE-USDC-SPOT");
        assert_eq!(instrument("@207").as_ref(), "HYPE-USDT0-SPOT");
    }

    #[test]
    fn unknown_coins_are_recorded_once_and_known_ones_never() {
        let mut unknown = UnknownCoins::default();
        for coin in ["#12", "#12", "BTC", "@107", "PURR/USDC", "+3"] {
            unknown.warn_once(coin);
        }
        assert_eq!(
            unknown.0,
            HashSet::from(["#12".to_owned(), "+3".to_owned()])
        );
    }

    /// [`info_tests::cancel_answer`] as the SDK reads it.
    fn cancel_answer(statuses: serde_json::Value) -> ExchangeResponseStatus {
        serde_json::from_value(info_tests::cancel_answer(statuses)).unwrap()
    }

    #[test]
    fn only_a_success_status_is_a_cancel() {
        assert_eq!(
            cancel_outcome(cancel_answer(serde_json::json!(["success"]))),
            Ok(())
        );
    }

    /// The exchange endpoint documentation's own example of a cancel that was not applied.
    #[test]
    fn an_error_status_under_a_top_level_ok_is_not_a_cancel() {
        let answer = cancel_answer(serde_json::json!([
            {"error": info_tests::CANCEL_NOT_APPLIED}
        ]));
        assert_eq!(
            cancel_outcome(answer),
            Err(info_tests::CANCEL_NOT_APPLIED.to_owned())
        );
    }

    #[test]
    fn an_answer_without_a_success_status_is_not_a_cancel() {
        let no_data: ExchangeResponseStatus = serde_json::from_value(serde_json::json!({
            "status": "ok",
            "response": {"type": "cancel"},
        }))
        .unwrap();
        let answers = [
            no_data,
            cancel_answer(serde_json::json!([])),
            cancel_answer(serde_json::json!([{"resting": {"oid": 1}}])),
        ];
        for answer in answers {
            let outcome = cancel_outcome(answer.clone());
            assert!(outcome.is_err(), "{answer:?} read as {outcome:?}");
        }
    }

    #[test]
    fn a_top_level_error_is_not_a_cancel() {
        let answer: ExchangeResponseStatus = serde_json::from_value(serde_json::json!({
            "status": "err",
            "response": "User or API Wallet does not exist.",
        }))
        .unwrap();
        assert_eq!(
            cancel_outcome(answer),
            Err("User or API Wallet does not exist.".to_owned())
        );
    }

    #[test]
    fn test_instrument_to_spot_coin() {
        let coin = instrument_to_spot_coin(&InstrumentNameExchange::from("PURR-USDC-SPOT"));
        assert_eq!(coin, Some("PURR/USDC".to_string()));

        let coin = instrument_to_spot_coin(&InstrumentNameExchange::from("HYPE-USDC-SPOT"));
        assert_eq!(coin, Some("HYPE/USDC".to_string()));

        // Malformed instruments return None
        assert_eq!(
            instrument_to_spot_coin(&InstrumentNameExchange::from("BTC-USDC-PERP")),
            None
        );
        assert_eq!(
            instrument_to_spot_coin(&InstrumentNameExchange::from("INVALID")),
            None
        );
    }

    #[test]
    fn test_map_tif() {
        assert_eq!(
            map_tif(&TimeInForce::GoodUntilCancelled { post_only: false }),
            "Gtc"
        );
        assert_eq!(
            map_tif(&TimeInForce::GoodUntilCancelled { post_only: true }),
            "Alo"
        );
        assert_eq!(map_tif(&TimeInForce::ImmediateOrCancel), "Ioc");
        assert_eq!(map_tif(&TimeInForce::FillOrKill), "Ioc");
        assert_eq!(map_tif(&TimeInForce::GoodUntilEndOfDay), "Gtc");
    }

    #[test]
    fn test_millis_to_datetime() {
        let dt = millis_to_datetime(1714100000000).unwrap();
        assert_eq!(dt.timestamp_millis(), 1714100000000);

        // Zero timestamp (Unix epoch) is valid
        assert!(millis_to_datetime(0).is_some());
    }
}

#[cfg(test)]
// Test code: panics on bad input are acceptable
#[allow(clippy::unwrap_used)]
pub(super) mod info_tests {
    use super::*;
    use crate::error::ApiError;
    use rust_decimal_macros::dec;
    use wiremock::matchers::{method, path};
    use wiremock::{Mock, MockServer, ResponseTemplate};

    /// One fill exactly as the `userFills` documentation shows it, including the two fields the
    /// SDK's own response type drops.
    const DOCUMENTED_FILL: &str = r#"{
        "closedPnl": "0.0",
        "coin": "AVAX",
        "crossed": false,
        "dir": "Open Long",
        "hash": "0xa166e3fa63c25663024b03f2e0da011a00307e4017465df020210d3d432e7cb8",
        "oid": 90542681,
        "px": "18.435",
        "side": "B",
        "startPosition": "26.86",
        "sz": "93.53",
        "time": 1681222254710,
        "fee": "0.01",
        "feeToken": "USDC",
        "builderFee": "0.01",
        "tid": 118906512037719
    }"#;

    /// Build an `InfoClient` pointed at `uri`. `InfoClient::new` opens no connection -- it only
    /// fills in the struct -- so this costs nothing and touches no network.
    pub(in crate::client::hyperliquid) async fn info_client_against(
        uri: String,
    ) -> hyperliquid_rust_sdk::InfoClient {
        let mut client = hyperliquid_rust_sdk::InfoClient::new(
            None,
            Some(hyperliquid_rust_sdk::BaseUrl::Localhost),
        )
        .await
        .unwrap();
        client.http_client.base_url = uri;
        client
    }

    /// The wallet clients built in tests sign with. No real server sees what it signs.
    pub(in crate::client::hyperliquid) fn test_wallet() -> ethers::signers::LocalWallet {
        "ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80"
            .parse()
            .unwrap()
    }

    /// Build an `ExchangeClient` pointed at `uri` that knows the assets in `coin_to_asset`.
    /// `ExchangeClient::new` would read them over the network instead.
    pub(in crate::client::hyperliquid) async fn exchange_client_against(
        uri: String,
        coin_to_asset: HashMap<String, u32>,
    ) -> hyperliquid_rust_sdk::ExchangeClient {
        hyperliquid_rust_sdk::ExchangeClient {
            http_client: info_client_against(uri).await.http_client,
            wallet: test_wallet(),
            meta: hyperliquid_rust_sdk::Meta {
                universe: Vec::new(),
            },
            vault_address: None,
            coin_to_asset,
        }
    }

    /// Hyperliquid's answer to a cancel, as its exchange endpoint sends it, carrying `statuses`.
    pub(in crate::client::hyperliquid) fn cancel_answer(
        statuses: serde_json::Value,
    ) -> serde_json::Value {
        serde_json::json!({
            "status": "ok",
            "response": {"type": "cancel", "data": {"statuses": statuses}},
        })
    }

    /// The documentation's example of a cancel Hyperliquid did not apply, under a top-level `ok`.
    pub(in crate::client::hyperliquid) const CANCEL_NOT_APPLIED: &str =
        "Order was never placed, already canceled, or filled.";

    async fn serve(body: String) -> (MockServer, hyperliquid_rust_sdk::InfoClient) {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/info"))
            .respond_with(ResponseTemplate::new(200).set_body_raw(body, "application/json"))
            .mount(&server)
            .await;
        let client = info_client_against(server.uri()).await;
        (server, client)
    }

    #[tokio::test]
    async fn a_fill_carries_the_tid_and_fee_token_the_sdk_type_drops() {
        let (_server, client) = serve(format!("[{DOCUMENTED_FILL}]")).await;

        let fills = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap();

        assert_eq!(fills.len(), 1);
        assert_eq!(fills[0].tid, 118_906_512_037_719);
        assert_eq!(fills[0].fee_token.as_deref(), Some("USDC"));
        assert_eq!(fills[0].oid, 90_542_681);
        assert_eq!(fills[0].coin, "AVAX");
    }

    #[tokio::test]
    async fn one_transaction_can_carry_several_fills_and_they_are_told_apart_by_tid() {
        // A single aggressive order sweeping three resting orders: one `hash`, one `oid`, three
        // fills. This is the shape the old `TradeId(hash)` collapsed into one trade.
        let sweep = r#"[
            {"closedPnl":"0.0","coin":"AVAX","crossed":true,"dir":"Open Long",
             "hash":"0xsweep","oid":1,"px":"18.40","side":"B","startPosition":"0",
             "sz":"10","time":1681222254710,"fee":"0.01","feeToken":"USDC","tid":111},
            {"closedPnl":"0.0","coin":"AVAX","crossed":true,"dir":"Open Long",
             "hash":"0xsweep","oid":1,"px":"18.45","side":"B","startPosition":"10",
             "sz":"10","time":1681222254710,"fee":"0.01","feeToken":"USDC","tid":222},
            {"closedPnl":"0.0","coin":"AVAX","crossed":true,"dir":"Open Long",
             "hash":"0xsweep","oid":1,"px":"18.50","side":"B","startPosition":"20",
             "sz":"10","time":1681222254710,"fee":"0.01","feeToken":"USDC","tid":333}
        ]"#;
        let (_server, client) = serve(sweep.to_string()).await;

        let fills = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap();

        let tids: Vec<u64> = fills.iter().map(|f| f.tid).collect();
        assert_eq!(tids, vec![111, 222, 333]);
    }

    #[tokio::test]
    async fn a_missing_fee_token_falls_back_rather_than_failing() {
        // `feeToken` is the one field here that nothing proves the endpoint sends, so its absence
        // must not cost the caller the whole response.
        let without = DOCUMENTED_FILL.replace(r#""feeToken": "USDC","#, "");
        let (_server, client) = serve(format!("[{without}]")).await;

        let fills = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap();

        assert_eq!(fills.len(), 1);
        assert!(fills[0].fee_token.is_none());
    }

    #[tokio::test]
    async fn a_missing_tid_fails_loudly_instead_of_falling_back_to_the_hash() {
        // Silently reverting to `hash` would reinstate the duplicate-id defect, and it would only
        // become visible as fills going missing during reconciliation.
        let without = DOCUMENTED_FILL.replace(r#""tid": 118906512037719"#, r#""unused": 0"#);
        let (_server, client) = serve(format!("[{without}]")).await;

        let error = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap_err();

        assert!(
            matches!(&error, UnindexedClientError::Internal(msg) if msg.contains("userFills")),
            "expected a parse failure naming the endpoint, got {error:?}"
        );
    }

    #[tokio::test]
    async fn the_request_names_the_address_in_full() {
        // `Display` for H160 abbreviates the middle of an address, which the venue would not
        // recognise. Only the `Debug` rendering is the full 0x-prefixed form.
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/info"))
            .and(wiremock::matchers::body_string_contains(
                "0x000000000000000000000000000000000000dead",
            ))
            .and(wiremock::matchers::body_string_contains("userFills"))
            .respond_with(ResponseTemplate::new(200).set_body_raw("[]", "application/json"))
            .mount(&server)
            .await;
        let client = info_client_against(server.uri()).await;

        let address = "0x000000000000000000000000000000000000dEaD"
            .parse::<ethers::types::H160>()
            .unwrap();

        // The mock only answers a body carrying the full address, so reaching `Ok` is the
        // assertion.
        assert!(
            user_fills_by_time(&client, address, 0, u64::MAX)
                .await
                .unwrap()
                .is_empty()
        );
    }

    /// `userFillsByTime` over `fills`, which are sorted by time, as observed on mainnet: both
    /// bounds inclusive, oldest first, at most [`USER_FILLS_PER_RESPONSE`] per response.
    struct FillsByTime {
        fills: Vec<serde_json::Value>,
    }

    impl wiremock::Respond for FillsByTime {
        fn respond(&self, request: &wiremock::Request) -> ResponseTemplate {
            let body: serde_json::Value = serde_json::from_slice(&request.body).unwrap();
            assert_eq!(body["type"], "userFillsByTime");
            let bound = |name: &str| body[name].as_u64().unwrap();
            let (start, end) = (bound("startTime"), bound("endTime"));
            let page: Vec<_> = self
                .fills
                .iter()
                .filter(|fill| (start..=end).contains(&fill["time"].as_u64().unwrap()))
                .take(USER_FILLS_PER_RESPONSE)
                .cloned()
                .collect();
            ResponseTemplate::new(200).set_body_json(page)
        }
    }

    /// `DOCUMENTED_FILL` with `tid` at `time`.
    fn fill(tid: u64, time: u64) -> serde_json::Value {
        let mut fill: serde_json::Value = serde_json::from_str(DOCUMENTED_FILL).unwrap();
        fill["tid"] = tid.into();
        fill["time"] = time.into();
        fill
    }

    async fn serve_fills_by_time(
        fills: Vec<serde_json::Value>,
    ) -> (MockServer, hyperliquid_rust_sdk::InfoClient) {
        let server = MockServer::start().await;
        Mock::given(method("POST"))
            .and(path("/info"))
            .respond_with(FillsByTime { fills })
            .mount(&server)
            .await;
        let client = info_client_against(server.uri()).await;
        (server, client)
    }

    /// A span holding more fills than one response is read completely, each fill once. Three
    /// fills share each millisecond, so a response ends inside one, and the next, read from that
    /// millisecond, returns again the fills of it already read.
    #[tokio::test]
    async fn a_span_with_more_fills_than_one_response_is_read_whole() {
        let fills = (0..4_500).map(|tid| fill(tid, 1_000 + tid / 3)).collect();
        let (server, client) = serve_fills_by_time(fills).await;

        let read = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap();
        let tids: Vec<u64> = read.iter().map(|fill| fill.tid).collect();
        assert_eq!(tids, (0..4_500).collect::<Vec<_>>());
        assert_eq!(server.received_requests().await.map(|r| r.len()), Some(3));

        // Both bounds are inclusive.
        let read = user_fills_by_time(&client, ethers::types::H160::zero(), 1_010, 1_019)
            .await
            .unwrap();
        assert_eq!(read.len(), 30);
    }

    /// A full response of fills all at the millisecond it was read from cannot be read past.
    #[tokio::test]
    async fn a_read_that_cannot_advance_is_truncated() {
        let fills = (0..USER_FILLS_PER_RESPONSE as u64)
            .map(|tid| fill(tid, 5))
            .collect();
        let (_server, client) = serve_fills_by_time(fills).await;

        let error = user_fills_by_time(&client, ethers::types::H160::zero(), 0, u64::MAX)
            .await
            .unwrap_err();
        assert!(
            matches!(error, UnindexedClientError::Truncated { fills_read: 2_000 }),
            "got {error:?}"
        );
    }

    #[test]
    fn a_span_is_read_in_whole_milliseconds_and_an_empty_one_not_at_all() {
        let at = |s: &str| s.parse::<DateTime<Utc>>().unwrap();
        assert_eq!(
            span_millis(
                at("1970-01-01T00:00:01.000900Z"),
                at("1970-01-01T00:00:02.000900Z")
            ),
            Some((1_000, 2_000))
        );
        assert_eq!(
            span_millis(at("1969-12-31T23:59:59Z"), at("1970-01-01T00:00:01Z")),
            Some((0, 1_000))
        );
        assert_eq!(
            span_millis(at("1970-01-01T00:00:02Z"), at("1970-01-01T00:00:01Z")),
            None
        );
        assert_eq!(
            span_millis(at("1969-01-01T00:00:00Z"), at("1969-12-31T23:59:59Z")),
            None
        );
    }

    // ---- client order ids -------------------------------------------------------------------

    const CID: &str = "85b60bff-64d5-49ec-95af-6268d7d1df63";
    /// `CID` as the venue reports it back: `0x` and the 32 lowercase hex digits.
    const CLOID: &str = "0x85b60bff64d549ec95af6268d7d1df63";

    /// A REST fill decodes its order's cloid, and one placed without a cloid decodes as `None`.
    #[test]
    fn a_user_fill_reads_its_cloid() {
        let fill = |cloid: &str| {
            serde_json::from_str::<UserFill>(&format!(
                r#"{{"coin": "BTC", "side": "B", "px": "1", "sz": "1", "time": 1, "oid": 7,
                    "fee": "0", "tid": 9{cloid}}}"#
            ))
            .unwrap()
        };

        let with = fill(&format!(r#", "cloid": "{CLOID}""#));
        assert_eq!(with.cloid.as_deref(), Some(CLOID));
        assert!(cloid_to_cid(CLOID).is_some());
        assert_eq!(fill("").cloid, None);
        assert_eq!(fill(r#", "cloid": null"#).cloid, None);
    }

    #[test]
    fn only_the_canonical_uuid_form_can_be_a_cloid() {
        let cloid = cid_to_cloid(&ClientOrderId::new(CID)).unwrap();
        assert_eq!(cloid.to_string(), CID);

        // Every other spelling of the same UUID would come back as `CID`, not as itself.
        for other in [
            "85B60BFF-64D5-49EC-95AF-6268D7D1DF63",
            "85b60bff64d549ec95af6268d7d1df63",
            "{85b60bff-64d5-49ec-95af-6268d7d1df63}",
            "urn:uuid:85b60bff-64d5-49ec-95af-6268d7d1df63",
            "example-1714100000000",
            "",
        ] {
            assert_eq!(cid_to_cloid(&ClientOrderId::new(other)), None, "{other}");
        }
    }

    #[test]
    fn a_generated_uuid_client_id_is_always_a_cloid() {
        for _ in 0..32 {
            let cid = ClientOrderId::uuid();
            assert!(cid_to_cloid(&cid).is_some(), "{cid}");
        }
    }

    #[test]
    fn the_reported_cloid_is_the_client_id_the_order_was_placed_under() {
        assert_eq!(cloid_to_cid(CLOID), Some(ClientOrderId::new(CID)));

        // The round trip through the form the SDK sends, for a fresh id.
        let cid = ClientOrderId::uuid();
        let sent = format!("0x{}", cid_to_cloid(&cid).unwrap().simple());
        assert_eq!(cloid_to_cid(&sent), Some(cid));
    }

    #[test]
    fn a_cloid_not_in_the_venue_form_names_no_client_id() {
        for malformed in [
            CID,
            "85b60bff64d549ec95af6268d7d1df63",
            "0x85b60bff64d549ec95af6268d7d1df",
            "0x85b60bff64d549ec95af6268d7d1df6300",
            "0x85b60bff-64d5-49ec-95af-6268d7d1df63",
            "0xzzb60bff64d549ec95af6268d7d1df63",
            "0x",
        ] {
            assert_eq!(cloid_to_cid(malformed), None, "{malformed}");
        }
    }

    #[test]
    fn a_record_without_a_cloid_is_reported_under_its_oid() {
        assert_eq!(record_cid(None, 42), Some(ClientOrderId::new("42")));
        assert_eq!(record_cid(Some(CLOID), 42), Some(ClientOrderId::new(CID)));
        assert_eq!(record_cid(Some("0x1234"), 42), None);
    }

    // ---- order statuses -----------------------------------------------------------------------

    #[test]
    fn every_documented_order_status_is_classified() {
        use OrderStatus::*;
        for (status, expected) in [
            ("open", Open),
            ("triggered", Open),
            ("filled", Filled),
            ("canceled", Cancelled),
            ("marginCanceled", Cancelled),
            ("vaultWithdrawalCanceled", Cancelled),
            ("openInterestCapCanceled", Cancelled),
            ("selfTradeCanceled", Cancelled),
            ("reduceOnlyCanceled", Cancelled),
            ("siblingFilledCanceled", Cancelled),
            ("delistedCanceled", Cancelled),
            ("liquidatedCanceled", Cancelled),
            ("scheduledCancel", Cancelled),
            ("rejected", Rejected),
            ("tickRejected", Rejected),
            ("minTradeNtlRejected", Rejected),
            ("perpMarginRejected", Rejected),
            ("reduceOnlyRejected", Rejected),
            ("badAloPxRejected", Rejected),
            ("iocCancelRejected", Rejected),
            ("badTriggerPxRejected", Rejected),
            ("marketOrderNoLiquidityRejected", Rejected),
            ("positionIncreaseAtOpenInterestCapRejected", Rejected),
            ("positionFlipAtOpenInterestCapRejected", Rejected),
            ("tooAggressiveAtOpenInterestCapRejected", Rejected),
            ("openInterestIncreaseRejected", Rejected),
            ("insufficientSpotBalanceRejected", Rejected),
            ("oracleRejected", Rejected),
            ("perpMaxPositionRejected", Rejected),
        ] {
            assert_eq!(OrderStatus::classify(status), Some(expected), "{status}");
        }
        assert_eq!(OrderStatus::classify("unknown_status"), None);
    }

    fn order_update(status: &str, cloid: &str, sz: &str) -> hyperliquid_rust_sdk::OrderUpdate {
        serde_json::from_str(&format!(
            r#"{{"order": {{"coin": "ETH", "side": "B", "limitPx": "1605.0", "sz": "{sz}",
                 "oid": 7001, "timestamp": 1714100000000, "origSz": "0.0075", "cloid": {cloid}}},
               "status": "{status}", "statusTimestamp": 1714100001000}}"#
        ))
        .unwrap()
    }

    fn snapshot_of(event: UnindexedAccountEvent) -> UnindexedOrderSnapshot {
        match event.kind {
            crate::AccountEventKind::OrderSnapshot(snapshot) => snapshot.0,
            other => panic!("expected an order snapshot, got {other:?}"),
        }
    }

    fn eth() -> InstrumentNameExchange {
        InstrumentNameExchange::from("ETH-USDC-PERP")
    }

    #[test]
    fn an_order_update_names_the_order_by_the_id_it_was_placed_under() {
        let update = order_update("open", &format!(r#""{CLOID}""#), "0.005");
        let order = snapshot_of(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
        );

        assert_eq!(order.key.cid, ClientOrderId::new(CID));
        assert_eq!(order.quantity, dec!(0.0075));
        match order.state {
            OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) => {
                assert_eq!(open.id, VenueOrderId::Assigned(OrderId::new("7001")));
                assert_eq!(open.filled_quantity, dec!(0.0025));
            }
            other => panic!("expected Open, got {other:?}"),
        }
    }

    #[test]
    fn an_order_update_without_a_cloid_is_named_by_its_oid() {
        let update = order_update("open", "null", "0.0075");
        let order = snapshot_of(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
        );
        assert_eq!(order.key.cid, ClientOrderId::new("7001"));
    }

    #[test]
    fn an_order_update_with_a_malformed_cloid_is_dropped() {
        let update = order_update("open", r#""0x1234""#, "0.0075");
        assert!(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).is_none()
        );
    }

    #[test]
    fn a_cancellation_for_any_reason_ends_the_order() {
        let update = order_update("selfTradeCanceled", &format!(r#""{CLOID}""#), "0.005");
        let order = snapshot_of(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
        );
        match order.state {
            OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(cancelled)) => {
                assert_eq!(cancelled.filled_quantity, Some(dec!(0.0025)));
            }
            other => panic!("expected Cancelled, got {other:?}"),
        }
    }

    /// A cancellation whose remaining size does not parse still ends the order, with its fill
    /// unknown rather than the update dropped.
    #[test]
    fn a_cancellation_with_an_unparseable_size_ends_the_order_with_its_fill_unknown() {
        let update = order_update("canceled", &format!(r#""{CLOID}""#), "not a number");
        let order = snapshot_of(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
        );
        match order.state {
            OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(cancelled)) => {
                assert_eq!(cancelled.filled_quantity, None);
            }
            other => panic!("expected Cancelled, got {other:?}"),
        }
    }

    #[test]
    fn a_status_saying_funds_ran_short_ends_the_order_as_balance_insufficient() {
        for status in ["perpMarginRejected", "insufficientSpotBalanceRejected"] {
            let update = order_update(status, &format!(r#""{CLOID}""#), "0.0075");
            let order = snapshot_of(
                order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
            );
            match order.state {
                OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
                    OrderError::Rejected(ApiError::BalanceInsufficient(None, reason)),
                )) => assert_eq!(reason, status),
                other => panic!("{status}: expected BalanceInsufficient, got {other:?}"),
            }
        }
    }

    #[test]
    fn a_rejection_ends_the_order_and_names_the_reason() {
        let update = order_update("badAloPxRejected", &format!(r#""{CLOID}""#), "0.0075");
        let order = snapshot_of(
            order_update_to_account_event(&update, ExchangeId::HyperliquidPerp, eth()).unwrap(),
        );
        match order.state {
            OrderState::Inactive(crate::order::state::InactiveOrderState::OpenFailed(
                OrderError::Rejected(ApiError::OrderRejected(reason)),
            )) => assert_eq!(reason, "badAloPxRejected"),
            other => panic!("expected a rejection, got {other:?}"),
        }
    }

    // ---- open orders --------------------------------------------------------------------------

    /// The shape `openOrders` was observed to return: `origSz` and `cloid` included, although the
    /// endpoint's documentation lists neither. Values are synthetic.
    const OPEN_ORDERS: &str = r#"[
        {"coin":"ETH","side":"B","limitPx":"1605.0","sz":"0.005","oid":7001,
         "timestamp":1714100000000,"origSz":"0.0075","cloid":"0x85b60bff64d549ec95af6268d7d1df63"},
        {"coin":"ETH","side":"A","limitPx":"1700.0","sz":"0.01","oid":7002,
         "timestamp":1714100000000,"origSz":"0.01"},
        {"coin":"BTC","side":"A","limitPx":"50000.0","sz":"0.001","oid":7003,
         "timestamp":1714100000000}
    ]"#;

    #[tokio::test]
    async fn open_orders_carry_the_cloid_and_original_size_the_sdk_type_drops() {
        let (_server, client) = serve(OPEN_ORDERS.to_string()).await;

        let rows = open_orders(&client, ethers::types::H160::zero())
            .await
            .unwrap();

        assert_eq!(rows.len(), 3);
        assert_eq!(rows[0].cloid.as_deref(), Some(CLOID));
        assert_eq!(rows[0].orig_sz.as_deref(), Some("0.0075"));
        assert_eq!(rows[1].cloid, None);
        assert_eq!(rows[2].orig_sz, None);
    }

    fn rows() -> Vec<OpenOrder> {
        serde_json::from_str(OPEN_ORDERS).unwrap()
    }

    fn perp(coin: &str) -> Option<InstrumentNameExchange> {
        Some(InstrumentNameExchange::from(format!("{coin}-USDC-PERP")))
    }

    #[test]
    fn an_open_order_is_listed_under_its_client_id_with_its_fill_so_far() {
        let order = open_order_to_order(&rows()[0], ExchangeId::HyperliquidPerp, eth()).unwrap();

        assert_eq!(order.key.cid, ClientOrderId::new(CID));
        assert_eq!(order.quantity, dec!(0.0075));
        assert_eq!(order.state.filled_quantity, dec!(0.0025));
        assert_eq!(order.state.id, VenueOrderId::Assigned(OrderId::new("7001")));
    }

    #[test]
    fn an_open_order_without_its_original_size_is_its_remainder_with_nothing_filled() {
        let row = &rows()[2];
        let btc = InstrumentNameExchange::from("BTC-USDC-PERP");
        let order = open_order_to_order(row, ExchangeId::HyperliquidPerp, btc).unwrap();

        assert_eq!(order.key.cid, ClientOrderId::new("7003"));
        assert_eq!(order.quantity, dec!(0.001));
        assert_eq!(order.state.filled_quantity, Decimal::ZERO);
    }

    #[test]
    fn every_requested_instrument_is_listed_complete_even_with_nothing_open() {
        let sol = InstrumentNameExchange::from("SOL-USDC-PERP");
        let mut listing = OpenOrderListing::new(
            &rows(),
            ExchangeId::HyperliquidPerp,
            &[eth(), sol.clone()],
            perp,
        );

        let eth_entry = listing.remove(&eth()).unwrap();
        assert!(eth_entry.orders_complete);
        assert_eq!(eth_entry.orders.len(), 2);

        // Nothing open: the entry is what lets an order that ended offline be seen to be gone.
        let sol_entry = listing.remove(&sol).unwrap();
        assert!(sol_entry.orders_complete);
        assert!(sol_entry.orders.is_empty());

        // BTC was not requested.
        assert_eq!(listing.into_snapshots().count(), 0);
    }

    #[test]
    fn an_unfiltered_listing_covers_every_instrument_with_open_orders() {
        let listing = OpenOrderListing::new(&rows(), ExchangeId::HyperliquidPerp, &[], perp);
        let mut instruments = listing
            .into_snapshots()
            .map(|entry| entry.instrument.to_string())
            .collect::<Vec<_>>();
        instruments.sort();
        assert_eq!(instruments, ["BTC-USDC-PERP", "ETH-USDC-PERP"]);
    }

    #[test]
    fn an_order_that_does_not_convert_leaves_only_its_instrument_incomplete() {
        let mut rows = rows();
        rows[1].side = "?".to_string();
        let btc = InstrumentNameExchange::from("BTC-USDC-PERP");
        let mut listing = OpenOrderListing::new(
            &rows,
            ExchangeId::HyperliquidPerp,
            &[eth(), btc.clone()],
            perp,
        );

        let eth_entry = listing.remove(&eth()).unwrap();
        assert!(!eth_entry.orders_complete);
        assert_eq!(eth_entry.orders.len(), 1);
        assert!(listing.remove(&btc).unwrap().orders_complete);
    }

    #[test]
    fn an_order_with_a_malformed_cloid_leaves_its_instrument_incomplete() {
        let mut rows = rows();
        rows[0].cloid = Some("0x1234".to_string());
        let mut listing = OpenOrderListing::new(&rows, ExchangeId::HyperliquidPerp, &[eth()], perp);

        let eth_entry = listing.remove(&eth()).unwrap();
        assert!(!eth_entry.orders_complete);
        assert_eq!(eth_entry.orders.len(), 1);
    }

    #[test]
    fn coins_the_client_does_not_trade_are_left_out() {
        let pairs = super::super::spot_coins::test_spot_pairs();
        let listing = OpenOrderListing::new(&rows(), ExchangeId::HyperliquidSpot, &[], |coin| {
            pairs.get(coin).map(spot_pair_to_instrument)
        });
        assert_eq!(listing.into_snapshots().count(), 0);
    }
}