rustrade_execution/client/hyperliquid/

mod.rs

//! Hyperliquid ExecutionClient implementations for perpetual futures and spot trading.
//!
//! Uses the official `hyperliquid_rust_sdk` crate for REST and WebSocket API access.
//! Gated behind the "hyperliquid" feature flag.
//!
//! # Modules
//!
//! - [`HyperliquidClient`]: Perpetual futures client
//! - [`spot::HyperliquidSpotClient`]: Spot trading client
//!
//! # Authentication
//!
//! Hyperliquid uses EVM-based authentication (Ethereum private key + EIP-712 signatures)
//! instead of traditional API key/secret. The SDK handles all signing internally.
//!
//! # Architecture
//!
//! - REST (`InfoClient`): account_snapshot, fetch_balances, fetch_open_orders, fetch_trades.
//!   `fetch_trades` posts to `/info` through the SDK's HTTP client but parses the response into
//!   [`common::UserFill`] rather than the SDK's type, which drops `tid` and `feeToken`.
//! - REST (`ExchangeClient`): open_order, cancel_order
//! - WebSocket (`InfoClient` with `with_reconnect`): account_stream via UserFills + OrderUpdates subscriptions
//!
//! # SDK Delegation Model
//!
//! This client delegates connection management to the official `hyperliquid_rust_sdk`:
//!
//! | Responsibility | Implementation |
//! |----------------|----------------|
//! | **Reconnection** | `InfoClient::with_reconnect()` handles WebSocket reconnection automatically |
//! | **Heartbeat** | SDK manages WebSocket ping/pong internally |
//! | **Deduplication** | This client's own, on the fills stream. See below. |
//! | **Fill recovery** | Not implemented — use [`ExecutionClient::fetch_trades`] after reconnect if needed |
//!
//! **Caller responsibilities**:
//! - If fill recovery is critical, monitor for reconnection events and call `fetch_trades()`
//! - REST clients (`InfoClient::new()`, `ExchangeClient::new()`) do NOT auto-reconnect;
//!   only WebSocket streams via `with_reconnect()` do
//!
//! # Trade identity and duplicate fills
//!
//! A [`Trade`]'s id is the venue's `tid`, which identifies the fill. It is deliberately **not**
//! `hash`, which identifies the transaction: one aggressive order sweeping several resting orders
//! produces several fills under a single hash, and keying on the hash makes them indistinguishable.
//! Hyperliquid documents `tid` as unique per fill but qualified by coin rather than globally, so
//! callers reconciling across instruments should qualify it the same way.
//!
//! The fills subscription opens with a snapshot of recent fills and the SDK resubscribes on every
//! reconnect, so the same fill is redelivered each time the socket comes back. A trade is a delta
//! the consumer accumulates, so this client deduplicates the stream itself — the cache is scoped to
//! one [`ExecutionClient::account_stream`] call and survives reconnects within it.
//!
//! Order updates are deliberately **not** deduplicated. They assert absolute state rather than a
//! delta, so a replayed one is idempotent while a dropped one could strand a consumer on stale
//! state.
//!
//! [`ExecutionClient::fetch_trades`] does not deduplicate: it answers the window it was asked for,
//! and its results reach the caller rather than the stream. A caller feeding both into one consumer
//! should expect overlap and reconcile on the trade id.
//!
//! # Conditional Orders (Stop, TakeProfit)
//!
//! Hyperliquid supports conditional (trigger) orders via the `ClientOrder::Trigger` variant.
//!
//! **Supported order kinds**:
//! - `Stop` → market order triggered at stop price (`tpsl: "sl"`, `is_market: true`)
//! - `StopLimit` → limit order triggered at stop price (`tpsl: "sl"`, `is_market: false`)
//! - `TakeProfit` → market order triggered at take-profit price (`tpsl: "tp"`, `is_market: true`)
//! - `TakeProfitLimit` → limit order triggered at take-profit price (`tpsl: "tp"`, `is_market: false`)
//!
//! **Unsupported**: `TrailingStop`, `TrailingStopLimit` (Hyperliquid does not support trailing stops)
//!
//! **UUID requirement**: Trigger orders MUST use [`ClientOrderId::uuid()`] for the client order ID.
//! The SDK's `cancel_by_cloid()` requires a valid UUID. Non-UUID client IDs will be rejected
//! at order placement with a clear error message.
//!
//! **SDK limitations** (hyperliquid_rust_sdk 0.6.x):
//! - `fetch_open_orders`: Returns `OpenOrdersResponse` which lacks trigger fields (`is_trigger`,
//!   `trigger_px`). All orders appear as `OrderKind::Limit`.
//! - `account_stream` (WebSocket): `OrderUpdate` uses `BasicOrder` which also lacks trigger fields.
//!   Downstream consumers should correlate orders by `cloid` and track `OrderKind` from the
//!   placement response.
//!
//! # Limitations
//!
//! - **Price precision**: Hyperliquid requires 5 significant figures for prices

pub mod common;
pub mod config;
pub mod error;
pub mod spot;

use crate::client::dedup::{dedup_key_from_event, is_duplicate, new_dedup_cache};
use crate::{
    AccountEvent, AccountEventKind, AccountSnapshot, InstrumentAccountSnapshot,
    UnindexedAccountEvent, UnindexedAccountSnapshot,
    balance::{AssetBalance, Balance},
    client::ExecutionClient,
    emit_stream_terminated,
    error::{
        ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
        UnindexedOrderError,
    },
    order::{
        Order, OrderKey, OrderKind, TimeInForce,
        id::{ClientOrderId, OrderId, StrategyId},
        request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
        state::{Cancelled, Filled, Open, OrderState, UnindexedOrderState},
    },
    position::Position,
    trade::{AssetFees, Trade, TradeId},
};
use chrono::{DateTime, Utc};
use common::{
    CancelOnDropStream, cid_to_cloid, instrument_to_perp_coin, map_tif, millis_to_datetime,
    parse_decimal, parse_side, perp_coin_to_instrument, round_to_5_sig_figs, user_fills,
};
pub use config::{HyperliquidConfig, HyperliquidConfigError};
use error::{map_order_error, map_sdk_error};
use ethers::signers::Signer;
use futures::{StreamExt, stream::BoxStream};
use hyperliquid_rust_sdk::{BaseUrl, ExchangeClient, InfoClient, Message, Subscription};
use rust_decimal::Decimal;
use rustrade_instrument::{
    Side,
    asset::name::AssetNameExchange,
    exchange::ExchangeId,
    instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
};
use rustrade_integration::collection::snapshot::Snapshot;
use smol_str::{SmolStr, format_smolstr};
use std::{
    collections::HashSet,
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, trace, warn};

/// USDC asset name on Hyperliquid (the only collateral asset for perps).
const USDC_ASSET: &str = "USDC";

/// Hyperliquid perpetual futures execution client.
///
/// Wraps the official `hyperliquid_rust_sdk` to implement the `ExecutionClient` trait.
/// Supports perpetual futures trading on Hyperliquid DEX.
#[derive(Debug, Clone)]
pub struct HyperliquidClient {
    config: HyperliquidConfig,
    info_client: Arc<InfoClient>,
    exchange_client: Arc<ExchangeClient>,
}

impl HyperliquidClient {
    /// Create a new client asynchronously.
    ///
    /// Use this when calling from an async context (e.g., tokio tests).
    /// For sync contexts, use `ExecutionClient::new()`.
    pub async fn connect(config: HyperliquidConfig) -> Result<Self, ConnectivityError> {
        let base_url = if config.testnet {
            BaseUrl::Testnet
        } else {
            BaseUrl::Mainnet
        };

        let info_client = InfoClient::new(None, Some(base_url))
            .await
            .map_err(|e| ConnectivityError::Socket(format!("InfoClient: {e}")))?;

        let wallet = config.wallet.clone();
        let exchange_client = ExchangeClient::new(None, wallet, Some(base_url), None, None)
            .await
            .map_err(|e| ConnectivityError::Socket(format!("ExchangeClient: {e}")))?;

        info!(
            testnet = config.testnet,
            wallet = %config.wallet_address_hex(),
            "Created HyperliquidClient"
        );

        Ok(Self {
            config,
            info_client: Arc::new(info_client),
            exchange_client: Arc::new(exchange_client),
        })
    }

    /// Returns the base URL for the configured network (mainnet or testnet).
    fn base_url(&self) -> BaseUrl {
        if self.config.testnet {
            BaseUrl::Testnet
        } else {
            BaseUrl::Mainnet
        }
    }

    /// Returns the wallet address as a hex string (for logging/debugging).
    pub fn wallet_address(&self) -> String {
        self.config.wallet_address_hex()
    }

    /// Returns the wallet address as ethers H160.
    fn wallet_h160(&self) -> ethers::types::H160 {
        self.config.wallet.address()
    }
}

impl ExecutionClient for HyperliquidClient {
    const EXCHANGE: ExchangeId = ExchangeId::HyperliquidPerp;

    // Perpetuals only — spot is served by `HyperliquidSpotClient`, which uses different coin
    // naming, asset indices and balance endpoints.
    const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] =
        &[InstrumentKindDiscriminant::Perpetual];

    type Config = HyperliquidConfig;
    type AccountStream = BoxStream<'static, UnindexedAccountEvent>;

    /// Creates a new Hyperliquid client synchronously.
    ///
    /// # Panics
    ///
    /// - If no Tokio runtime is available on the current thread
    /// - If called from within an async context (e.g., inside `async fn`, `spawn`, or `block_on`)
    /// - If SDK initialization fails (network error, invalid credentials)
    ///
    /// # Recommended Usage
    ///
    /// Use [`HyperliquidClient::connect`] instead — it's async-safe and returns `Result`.
    /// This method exists only for trait compliance; prefer `connect()` in all new code.
    fn new(config: Self::Config) -> Self {
        let base_url = if config.testnet {
            BaseUrl::Testnet
        } else {
            BaseUrl::Mainnet
        };

        // SDK initialization is async; block on it since ExecutionClient::new is sync.
        // WARNING: This will panic if called from within an async context.
        let handle = tokio::runtime::Handle::current();

        let info_client = handle.block_on(async {
            InfoClient::new(None, Some(base_url))
                .await
                .unwrap_or_else(|e| panic!("Failed to create Hyperliquid InfoClient: {e}"))
        });

        let wallet = config.wallet.clone();
        let exchange_client = handle.block_on(async {
            ExchangeClient::new(None, wallet, Some(base_url), None, None)
                .await
                .unwrap_or_else(|e| panic!("Failed to create Hyperliquid ExchangeClient: {e}"))
        });

        info!(
            testnet = config.testnet,
            wallet = %config.wallet_address_hex(),
            "Created HyperliquidClient"
        );

        Self {
            config,
            info_client: Arc::new(info_client),
            exchange_client: Arc::new(exchange_client),
        }
    }

    async fn account_snapshot(
        &self,
        _assets: &[AssetNameExchange],
        instruments: &[InstrumentNameExchange],
    ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
        let address = self.wallet_h160();

        // Fetch user state (balances + positions) and open orders concurrently
        let (user_state, open_orders) = tokio::try_join!(
            async {
                self.info_client
                    .user_state(address)
                    .await
                    .map_err(map_sdk_error)
            },
            async {
                self.info_client
                    .open_orders(address)
                    .await
                    .map_err(map_sdk_error)
            }
        )?;

        let now = Utc::now();

        // Build balance from margin summary (USDC is the only collateral)
        let account_value =
            parse_decimal(&user_state.margin_summary.account_value, "account_value")
                .unwrap_or(Decimal::ZERO);
        let margin_used = parse_decimal(
            &user_state.margin_summary.total_margin_used,
            "total_margin_used",
        )
        .unwrap_or(Decimal::ZERO);

        // Free balance can go negative during liquidation (margin_used > account_value).
        // Clamp to zero since negative free balance has no meaningful interpretation.
        let free_balance = (account_value - margin_used).max(Decimal::ZERO);
        let balances = vec![AssetBalance::new(
            AssetNameExchange::from(USDC_ASSET),
            Balance::new(account_value, free_balance),
            now,
        )];

        // Build instrument filter if provided
        let instrument_filter: Option<HashSet<_>> = if instruments.is_empty() {
            None
        } else {
            let mut set = HashSet::with_capacity(instruments.len());
            set.extend(instruments.iter().cloned());
            Some(set)
        };

        // Group open orders by instrument
        let mut orders_by_instrument: std::collections::HashMap<InstrumentNameExchange, Vec<_>> =
            std::collections::HashMap::with_capacity(open_orders.len());

        for order in &open_orders {
            let instrument = perp_coin_to_instrument(&order.coin);
            if instrument_filter
                .as_ref()
                .is_some_and(|f| !f.contains(&instrument))
            {
                continue;
            }

            let Some(side) = parse_side(&order.side) else {
                continue;
            };
            let Some(price) = parse_decimal(&order.limit_px, "limit_px") else {
                continue;
            };
            let Some(quantity) = parse_decimal(&order.sz, "sz") else {
                continue;
            };
            let Some(time_exchange) = millis_to_datetime(order.timestamp) else {
                warn!(
                    oid = order.oid,
                    timestamp = order.timestamp,
                    "Invalid order timestamp, skipping"
                );
                continue;
            };

            let order_id = format_smolstr!("{}", order.oid);
            let order_snapshot = Order {
                key: OrderKey {
                    exchange: ExchangeId::HyperliquidPerp,
                    instrument: instrument.clone(),
                    strategy: StrategyId::unknown(),
                    cid: ClientOrderId::new(order_id.clone()),
                },
                side,
                price: Some(price),
                quantity,
                kind: OrderKind::Limit,
                time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
                state: crate::order::state::OrderState::active(Open {
                    id: OrderId(order_id),
                    time_exchange,
                    filled_quantity: Decimal::ZERO,
                }),
            };

            orders_by_instrument
                .entry(instrument)
                .or_default()
                .push(order_snapshot);
        }

        // Build positions from asset_positions
        let mut instrument_snapshots = Vec::new();
        for asset_pos in user_state.asset_positions {
            let pos = &asset_pos.position;
            let instrument = perp_coin_to_instrument(&pos.coin);

            if instrument_filter
                .as_ref()
                .is_some_and(|f| !f.contains(&instrument))
            {
                continue;
            }

            let quantity = parse_decimal(&pos.szi, "szi").unwrap_or(Decimal::ZERO);
            let entry_price = pos
                .entry_px
                .as_ref()
                .and_then(|p| parse_decimal(p, "entry_px"));
            let unrealized_pnl = parse_decimal(&pos.unrealized_pnl, "unrealized_pnl");
            let margin_used = parse_decimal(&pos.margin_used, "margin_used");
            let liquidation_price = pos
                .liquidation_px
                .as_ref()
                .and_then(|p| parse_decimal(p, "liquidation_px"));
            let leverage = Some(Decimal::from(pos.leverage.value));

            let position = if quantity.is_zero() {
                None
            } else {
                Some(Position::new(
                    quantity,
                    entry_price,
                    unrealized_pnl,
                    margin_used,
                    liquidation_price,
                    leverage,
                    now,
                ))
            };

            let orders = orders_by_instrument.remove(&instrument).unwrap_or_default();

            instrument_snapshots.push(InstrumentAccountSnapshot {
                instrument,
                orders,
                position,
                isolated: None,
            });
        }

        // Add any instruments that have orders but no position
        for (instrument, orders) in orders_by_instrument {
            instrument_snapshots.push(InstrumentAccountSnapshot {
                instrument,
                orders,
                position: None,
                isolated: None,
            });
        }

        Ok(AccountSnapshot {
            exchange: ExchangeId::HyperliquidPerp,
            balances,
            instruments: instrument_snapshots,
        })
    }

    /// Returns a live stream of account events (fills, order updates).
    ///
    /// # Instrument filtering
    ///
    /// The `instruments` parameter is **ignored** — Hyperliquid's WebSocket API does not
    /// support per-instrument subscriptions for user events. All fills and order updates
    /// across all instruments are delivered. Consumers requiring instrument filtering
    /// must filter client-side.
    ///
    /// # Task lifecycle
    ///
    /// Spawns two background tasks (fills, orders) that are automatically cancelled
    /// when the returned stream is dropped. The `ws_client` is held by the orders task;
    /// when cancelled, both tasks exit and the WebSocket connection closes.
    async fn account_stream(
        &self,
        _assets: &[AssetNameExchange],
        _instruments: &[InstrumentNameExchange],
    ) -> Result<Self::AccountStream, UnindexedClientError> {
        let user = self.wallet_h160();
        let base_url = self.base_url();

        // Create a dedicated InfoClient for WebSocket streaming.
        // Using with_reconnect() enables SDK-managed reconnection.
        let mut ws_client = InfoClient::with_reconnect(None, Some(base_url))
            .await
            .map_err(|e| ConnectivityError::Socket(e.to_string()))?;

        // Create channels for subscriptions
        let (fills_tx, mut fills_rx) = mpsc::unbounded_channel::<Message>();
        let (orders_tx, mut orders_rx) = mpsc::unbounded_channel::<Message>();

        // Subscribe to user fills
        ws_client
            .subscribe(Subscription::UserFills { user }, fills_tx)
            .await
            .map_err(|e| ConnectivityError::Socket(format!("UserFills subscribe: {e}")))?;

        // Subscribe to order updates
        ws_client
            .subscribe(Subscription::OrderUpdates { user }, orders_tx)
            .await
            .map_err(|e| ConnectivityError::Socket(format!("OrderUpdates subscribe: {e}")))?;

        info!(%user, "Subscribed to Hyperliquid account stream");

        // Create output channel for merged events
        let (event_tx, event_rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();

        // CancellationToken ensures tasks exit when stream is dropped
        let cancel_token = CancellationToken::new();

        // Both tasks share `event_tx` and both observe `recv() -> None` when the SDK gives up on the
        // socket, so guard the terminal emit with a shared flag — only the first non-cancellation
        // terminal sends `StreamTerminated`, avoiding a double-emit. Relaxed ordering suffices: the
        // flag has no associated payload to synchronise; the channel send is the synchronisation point.
        let terminated = Arc::new(AtomicBool::new(false));

        // Scoped to this stream rather than to the client: a fresh `account_stream` opens a fresh
        // subscription, whose snapshot the new consumer has not seen. It must outlive a reconnect,
        // which it does -- the SDK reconnects underneath this task, not around it.
        let fills_dedup = new_dedup_cache();

        // Spawn task to process fills
        let fills_event_tx = event_tx.clone();
        let fills_cancel = cancel_token.clone();
        let fills_terminated = terminated.clone();
        tokio::spawn(async move {
            loop {
                tokio::select! {
                    biased;
                    () = fills_cancel.cancelled() => {
                        debug!("Fills task cancelled");
                        return;
                    }
                    msg = fills_rx.recv() => {
                        let Some(msg) = msg else {
                            debug!("Fills receiver closed");
                            // SDK gave up on the stream (channel closed). Emit a single terminal
                            // StreamTerminated across both tasks (guarded by the shared flag).
                            if fills_terminated
                                .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
                                .is_ok()
                            {
                                emit_stream_terminated(
                                    &fills_event_tx,
                                    ExchangeId::HyperliquidPerp,
                                    StreamTerminationReason::Error(
                                        "hyperliquid account stream closed".to_string(),
                                    ),
                                );
                            }
                            return;
                        };
                        match msg {
                            Message::UserFills(fills) => {
                                // Hyperliquid opens a `userFills` subscription with a snapshot of
                                // recent fills, and the SDK resubscribes on every reconnect. Each
                                // reconnect therefore redelivers fills already sent. A trade is a
                                // delta the consumer accumulates, so redelivering one double-counts
                                // filled quantity and fees -- hence the dedup cache.
                                //
                                // Order updates on the sibling task are deliberately NOT deduped:
                                // they are absolute state, so a replayed one is idempotent, while
                                // dropping one could strand the consumer on stale state.
                                for fill in fills.data.fills {
                                    let Some(event) = fill_to_account_event(&fill) else {
                                        continue;
                                    };
                                    if let Some(key) = dedup_key_from_event(&event)
                                        && is_duplicate(&fills_dedup, key)
                                    {
                                        trace!(
                                            tid = fill.tid,
                                            "Hyperliquid dedup: skipping fill already delivered"
                                        );
                                        continue;
                                    }
                                    if fills_event_tx.send(event).is_err() {
                                        debug!("Fills event channel closed");
                                        return;
                                    }
                                }
                            }
                            Message::NoData => {
                                warn!("UserFills WebSocket disconnected");
                            }
                            Message::HyperliquidError(e) => {
                                // Transient, non-terminal: the loop continues. Log only — no
                                // in-band event (consumers took no action on the old StreamError).
                                error!(%e, "UserFills WebSocket error");
                            }
                            _ => {}
                        }
                    }
                }
            }
        });

        // Spawn task to process order updates
        // NOTE: ws_client is moved here to keep the WebSocket alive. When this task
        // exits (via cancellation or channel close), the WebSocket connection closes,
        // which causes fills_rx to also close.
        let orders_event_tx = event_tx;
        let orders_cancel = cancel_token.clone();
        let orders_terminated = terminated;
        tokio::spawn(async move {
            let _ws_client = ws_client;

            loop {
                tokio::select! {
                    biased;
                    () = orders_cancel.cancelled() => {
                        debug!("Orders task cancelled");
                        return;
                    }
                    msg = orders_rx.recv() => {
                        let Some(msg) = msg else {
                            debug!("Orders receiver closed");
                            // SDK gave up on the stream (channel closed). Emit a single terminal
                            // StreamTerminated across both tasks (guarded by the shared flag).
                            if orders_terminated
                                .compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
                                .is_ok()
                            {
                                emit_stream_terminated(
                                    &orders_event_tx,
                                    ExchangeId::HyperliquidPerp,
                                    StreamTerminationReason::Error(
                                        "hyperliquid account stream closed".to_string(),
                                    ),
                                );
                            }
                            return;
                        };
                        match msg {
                            Message::OrderUpdates(updates) => {
                                for update in updates.data {
                                    if let Some(event) = order_update_to_account_event(&update)
                                        && orders_event_tx.send(event).is_err()
                                    {
                                        debug!("Orders event channel closed");
                                        return;
                                    }
                                }
                            }
                            Message::NoData => {
                                warn!("OrderUpdates WebSocket disconnected");
                            }
                            Message::HyperliquidError(e) => {
                                // Transient, non-terminal: the loop continues. Log only — no
                                // in-band event (consumers took no action on the old StreamError).
                                error!(%e, "OrderUpdates WebSocket error");
                            }
                            _ => {}
                        }
                    }
                }
            }
        });

        // Wrap stream with drop guard that cancels tasks
        let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(event_rx);
        let guarded_stream = CancelOnDropStream::new(stream, cancel_token);
        Ok(guarded_stream.boxed())
    }

    async fn cancel_order(
        &self,
        request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
    ) -> Option<UnindexedOrderResponseCancel> {
        use crate::order::{request::OrderResponseCancel, state::Cancelled};
        use hyperliquid_rust_sdk::{
            ClientCancelRequest, ClientCancelRequestCloid, ExchangeResponseStatus,
        };
        use uuid::Uuid;

        let coin = instrument_to_perp_coin(request.key.instrument);

        // Get order ID from request
        let order_id = match &request.state.id {
            Some(id) => id,
            None => {
                warn!("Cancel request missing order ID");
                return Some(OrderResponseCancel {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    state: Err(UnindexedOrderError::Rejected(
                        crate::error::ApiError::OrderRejected("Missing order ID".to_string()),
                    )),
                });
            }
        };

        // Try to determine cancel method:
        // 1. If order_id parses as u64, use cancel() with OID (regular limit orders)
        // 2. If order_id parses as UUID, use cancel_by_cloid() (trigger orders)
        // 3. Otherwise, reject with clear error
        enum CancelMethod {
            ByOid(u64),
            ByCloid(Uuid),
        }

        let cancel_method = if let Ok(oid) = order_id.0.parse::<u64>() {
            CancelMethod::ByOid(oid)
        } else if let Ok(uuid) = Uuid::parse_str(&order_id.0) {
            CancelMethod::ByCloid(uuid)
        } else {
            warn!(?order_id, "Order ID is neither u64 (OID) nor UUID (cloid)");
            return Some(OrderResponseCancel {
                key: OrderKey {
                    exchange: ExchangeId::HyperliquidPerp,
                    instrument: request.key.instrument.clone(),
                    strategy: request.key.strategy.clone(),
                    cid: request.key.cid.clone(),
                },
                state: Err(UnindexedOrderError::Rejected(
                    crate::error::ApiError::OrderRejected(
                        "Invalid order ID: must be numeric OID or UUID (for trigger orders)"
                            .to_string(),
                    ),
                )),
            });
        };

        let response = match cancel_method {
            CancelMethod::ByOid(oid) => {
                debug!(oid, "Cancelling by OID");
                let cancel_request = ClientCancelRequest { asset: coin, oid };
                self.exchange_client.cancel(cancel_request, None).await
            }
            CancelMethod::ByCloid(cloid) => {
                debug!(%cloid, "Cancelling by cloid (trigger order)");
                let cancel_request = ClientCancelRequestCloid { asset: coin, cloid };
                self.exchange_client
                    .cancel_by_cloid(cancel_request, None)
                    .await
            }
        };

        let response = match response {
            Ok(r) => r,
            Err(e) => {
                warn!(%e, "Cancel order failed (transport)");
                return Some(OrderResponseCancel {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    state: Err(map_order_error(e, request.key.instrument)),
                });
            }
        };

        match response {
            ExchangeResponseStatus::Ok(_) => {
                debug!("Cancel order accepted");
                // Hyperliquid cancel response doesn't include an exchange timestamp
                Some(OrderResponseCancel {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    state: Ok(Cancelled::new(
                        order_id.clone(),
                        Utc::now(),
                        Decimal::ZERO, // Cancel response doesn't include filled quantity
                    )),
                })
            }
            ExchangeResponseStatus::Err(msg) => {
                warn!(%msg, "Cancel rejected by exchange");
                Some(OrderResponseCancel {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    state: Err(UnindexedOrderError::Rejected(
                        crate::error::ApiError::OrderRejected(msg),
                    )),
                })
            }
        }
    }

    async fn open_order(
        &self,
        request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
    ) -> Option<Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>> {
        use hyperliquid_rust_sdk::{
            ClientLimit, ClientOrder, ClientOrderRequest, ClientTrigger, ExchangeDataStatus,
            ExchangeResponseStatus,
        };

        let make_rejected =
            |msg: String| -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
                Order {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    side: request.state.side,
                    price: request.state.price,
                    quantity: request.state.quantity,
                    kind: request.state.kind,
                    time_in_force: request.state.time_in_force,
                    state: OrderState::inactive(OrderError::Rejected(
                        crate::error::ApiError::OrderRejected(msg),
                    )),
                }
            };

        let make_unsupported =
            |msg: String| -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
                Order {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    side: request.state.side,
                    price: request.state.price,
                    quantity: request.state.quantity,
                    kind: request.state.kind,
                    time_in_force: request.state.time_in_force,
                    state: OrderState::inactive(OrderError::UnsupportedOrderType(msg)),
                }
            };

        let coin = instrument_to_perp_coin(request.key.instrument);
        let is_buy = request.state.side == Side::Buy;

        match request.state.kind {
            OrderKind::Market => {
                return Some(make_unsupported(
                    "Hyperliquid does not support market orders; use Limit with IOC time-in-force"
                        .to_string(),
                ));
            }
            OrderKind::TrailingStop { .. } | OrderKind::TrailingStopLimit { .. } => {
                return Some(make_unsupported(
                    "Hyperliquid does not support trailing stop orders".to_string(),
                ));
            }
            OrderKind::Limit
            | OrderKind::Stop { .. }
            | OrderKind::StopLimit { .. }
            | OrderKind::TakeProfit { .. }
            | OrderKind::TakeProfitLimit { .. } => {}
        }

        let cloid = cid_to_cloid(&request.key.cid);
        let cloid_is_some = cloid.is_some();
        let is_trigger_order = matches!(
            request.state.kind,
            OrderKind::Stop { .. }
                | OrderKind::StopLimit { .. }
                | OrderKind::TakeProfit { .. }
                | OrderKind::TakeProfitLimit { .. }
        );
        if is_trigger_order && cloid.is_none() {
            return Some(make_rejected(
                "Trigger orders require UUID-format client order ID (use ClientOrderId::uuid()). \
                 Non-UUID IDs cannot be cancelled via cancel_by_cloid()."
                    .to_string(),
            ));
        }

        let limit_px = match request.state.kind {
            // Market triggers: SDK uses trigger_px as limit_px
            OrderKind::Stop { trigger_price } | OrderKind::TakeProfit { trigger_price } => {
                round_to_5_sig_figs(trigger_price)
            }
            _ => match request.state.price {
                Some(p) => round_to_5_sig_figs(p),
                None => {
                    return Some(make_rejected(
                        "Hyperliquid requires limit price for Limit/StopLimit/TakeProfitLimit orders"
                            .to_string(),
                    ));
                }
            },
        };

        let sz = round_to_5_sig_figs(request.state.quantity);

        // Map time-in-force (warn if FOK is substituted with IOC)
        if matches!(request.state.time_in_force, TimeInForce::FillOrKill) {
            warn!(
                instrument = %request.key.instrument,
                "FillOrKill not supported by Hyperliquid, using ImmediateOrCancel (may result in partial fills)"
            );
        }
        let tif = map_tif(&request.state.time_in_force).to_string();

        // Build order_type based on OrderKind
        let order_type = match request.state.kind {
            OrderKind::Limit => ClientOrder::Limit(ClientLimit { tif }),
            OrderKind::Stop { trigger_price } => ClientOrder::Trigger(ClientTrigger {
                is_market: true,
                trigger_px: round_to_5_sig_figs(trigger_price),
                tpsl: "sl".to_string(),
            }),
            OrderKind::StopLimit { trigger_price } => ClientOrder::Trigger(ClientTrigger {
                is_market: false,
                trigger_px: round_to_5_sig_figs(trigger_price),
                tpsl: "sl".to_string(),
            }),
            OrderKind::TakeProfit { trigger_price } => ClientOrder::Trigger(ClientTrigger {
                is_market: true,
                trigger_px: round_to_5_sig_figs(trigger_price),
                tpsl: "tp".to_string(),
            }),
            OrderKind::TakeProfitLimit { trigger_price } => ClientOrder::Trigger(ClientTrigger {
                is_market: false,
                trigger_px: round_to_5_sig_figs(trigger_price),
                tpsl: "tp".to_string(),
            }),
            // Already rejected above
            OrderKind::Market
            | OrderKind::TrailingStop { .. }
            | OrderKind::TrailingStopLimit { .. } => {
                unreachable!("unsupported order kinds rejected earlier")
            }
        };

        let order_request = ClientOrderRequest {
            asset: coin,
            is_buy,
            reduce_only: request.state.reduce_only,
            limit_px,
            sz,
            cloid,
            order_type,
        };

        let response = match self.exchange_client.order(order_request, None).await {
            Ok(r) => r,
            Err(e) => {
                warn!(%e, "Open order failed");
                return Some(Order {
                    key: OrderKey {
                        exchange: ExchangeId::HyperliquidPerp,
                        instrument: request.key.instrument.clone(),
                        strategy: request.key.strategy.clone(),
                        cid: request.key.cid.clone(),
                    },
                    side: request.state.side,
                    price: request.state.price,
                    quantity: request.state.quantity,
                    kind: request.state.kind,
                    time_in_force: request.state.time_in_force,
                    state: OrderState::inactive(map_order_error(e, request.key.instrument)),
                });
            }
        };

        // Parse response
        let state = match response {
            ExchangeResponseStatus::Ok(exchange_resp) => {
                // Check status from response data
                let status = exchange_resp
                    .data
                    .and_then(|d| d.statuses.into_iter().next());

                match status {
                    Some(ExchangeDataStatus::Resting(resting)) => {
                        debug!(oid = resting.oid, "Order resting");
                        OrderState::active(Open {
                            id: OrderId(format_smolstr!("{}", resting.oid)),
                            time_exchange: Utc::now(),
                            filled_quantity: Decimal::ZERO,
                        })
                    }
                    Some(ExchangeDataStatus::Filled(filled)) => {
                        debug!(oid = filled.oid, avg_px = %filled.avg_px, "Order filled");
                        // Hyperliquid provides avg_px for filled orders
                        let avg_price = parse_decimal(&filled.avg_px, "avg_px");
                        OrderState::fully_filled(Filled::new(
                            OrderId(format_smolstr!("{}", filled.oid)),
                            Utc::now(),
                            parse_decimal(&filled.total_sz, "total_sz")
                                .unwrap_or(request.state.quantity),
                            avg_price,
                        ))
                    }
                    Some(ExchangeDataStatus::Error(msg)) => {
                        warn!(%msg, "Order rejected by exchange");
                        OrderState::inactive(OrderError::Rejected(
                            crate::error::ApiError::OrderRejected(msg),
                        ))
                    }
                    Some(
                        ExchangeDataStatus::WaitingForFill | ExchangeDataStatus::WaitingForTrigger,
                    ) => {
                        // Use cid as OrderId only if it's a valid UUID (required for
                        // cancel_by_cloid). Trigger orders pass the UUID guard above;
                        // limit IOC/FOK orders may have non-UUID cids — those become
                        // untrackable, so reject with an observable failure rather than
                        // returning an OrderId that cancel cannot parse.
                        if cloid_is_some {
                            debug!(cloid = %request.key.cid.0, "Order waiting (cloid trackable)");
                            OrderState::active(Open {
                                id: OrderId(request.key.cid.0.clone()),
                                time_exchange: Utc::now(),
                                filled_quantity: Decimal::ZERO,
                            })
                        } else {
                            warn!("Order accepted but cid is not UUID — untrackable");
                            OrderState::inactive(OrderError::Rejected(
                                crate::error::ApiError::OrderRejected(
                                    "order accepted but cid is not UUID; cannot track for cancel"
                                        .to_string(),
                                ),
                            ))
                        }
                    }
                    Some(ExchangeDataStatus::Success) | None => {
                        // Generic success without order ID — SDK didn't return structured data.
                        // This shouldn't happen for limit orders; reject to avoid silent failures.
                        warn!("Order accepted but no order ID returned");
                        OrderState::inactive(OrderError::Rejected(
                            crate::error::ApiError::OrderRejected(
                                "no order ID in response".to_string(),
                            ),
                        ))
                    }
                }
            }
            ExchangeResponseStatus::Err(msg) => {
                warn!(%msg, "Order rejected");
                OrderState::inactive(OrderError::Rejected(crate::error::ApiError::OrderRejected(
                    msg,
                )))
            }
        };

        Some(Order {
            key: OrderKey {
                exchange: ExchangeId::HyperliquidPerp,
                instrument: request.key.instrument.clone(),
                strategy: request.key.strategy.clone(),
                cid: request.key.cid.clone(),
            },
            side: request.state.side,
            price: request.state.price,
            quantity: request.state.quantity,
            kind: request.state.kind,
            time_in_force: request.state.time_in_force,
            state,
        })
    }

    async fn fetch_balances(
        &self,
        _assets: &[AssetNameExchange],
    ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
        let address = self.wallet_h160();

        let user_state = self
            .info_client
            .user_state(address)
            .await
            .map_err(map_sdk_error)?;

        let now = Utc::now();

        // Hyperliquid perps use USDC as the only collateral
        let account_value =
            parse_decimal(&user_state.margin_summary.account_value, "account_value")
                .unwrap_or(Decimal::ZERO);
        let margin_used = parse_decimal(
            &user_state.margin_summary.total_margin_used,
            "total_margin_used",
        )
        .unwrap_or(Decimal::ZERO);

        // Free balance can go negative during liquidation; clamp to zero
        let free_balance = (account_value - margin_used).max(Decimal::ZERO);
        Ok(vec![AssetBalance::new(
            AssetNameExchange::from(USDC_ASSET),
            Balance::new(account_value, free_balance),
            now,
        )])
    }

    async fn fetch_open_orders(
        &self,
        instruments: &[InstrumentNameExchange],
    ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
        let address = self.wallet_h160();

        let open_orders = self
            .info_client
            .open_orders(address)
            .await
            .map_err(map_sdk_error)?;

        let instrument_filter: Option<HashSet<_>> = if instruments.is_empty() {
            None
        } else {
            let mut set = HashSet::with_capacity(instruments.len());
            set.extend(instruments.iter().cloned());
            Some(set)
        };

        let mut result = Vec::new();
        for order in open_orders {
            let instrument = perp_coin_to_instrument(&order.coin);

            if instrument_filter
                .as_ref()
                .is_some_and(|f| !f.contains(&instrument))
            {
                continue;
            }

            let Some(side) = parse_side(&order.side) else {
                continue;
            };
            let Some(price) = parse_decimal(&order.limit_px, "limit_px") else {
                continue;
            };
            let Some(quantity) = parse_decimal(&order.sz, "sz") else {
                continue;
            };
            let Some(time_exchange) = millis_to_datetime(order.timestamp) else {
                warn!(
                    oid = order.oid,
                    timestamp = order.timestamp,
                    "Invalid order timestamp, skipping"
                );
                continue;
            };

            let order_id = format_smolstr!("{}", order.oid);
            result.push(Order {
                key: OrderKey {
                    exchange: ExchangeId::HyperliquidPerp,
                    instrument,
                    strategy: StrategyId::unknown(),
                    cid: ClientOrderId::new(order_id.clone()),
                },
                side,
                price: Some(price),
                quantity,
                kind: OrderKind::Limit,
                time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
                state: Open {
                    id: OrderId(order_id),
                    time_exchange,
                    filled_quantity: Decimal::ZERO,
                },
            });
        }

        Ok(result)
    }

    async fn fetch_trades(
        &self,
        time_since: DateTime<Utc>,
        instruments: &[InstrumentNameExchange],
    ) -> Result<Vec<Trade<AssetNameExchange, InstrumentNameExchange>>, UnindexedClientError> {
        let address = self.wallet_h160();

        let fills = user_fills(&self.info_client, address).await?;

        // Clamp to 0 for dates before epoch (shouldn't happen in practice)
        #[allow(clippy::cast_sign_loss)] // timestamp_millis >= 0 after max(0)
        let time_since_ms = time_since.timestamp_millis().max(0) as u64;

        let instrument_filter: Option<HashSet<_>> = if instruments.is_empty() {
            None
        } else {
            let mut set = HashSet::with_capacity(instruments.len());
            set.extend(instruments.iter().cloned());
            Some(set)
        };

        let mut result = Vec::new();
        for fill in fills {
            // Filter by time
            if fill.time < time_since_ms {
                continue;
            }

            let instrument = perp_coin_to_instrument(&fill.coin);

            if instrument_filter
                .as_ref()
                .is_some_and(|f| !f.contains(&instrument))
            {
                continue;
            }

            let Some(side) = parse_side(&fill.side) else {
                continue;
            };
            let Some(price) = parse_decimal(&fill.px, "px") else {
                continue;
            };
            let Some(quantity) = parse_decimal(&fill.sz, "sz") else {
                continue;
            };
            let fee = parse_decimal(&fill.fee, "fee").unwrap_or(Decimal::ZERO);

            let Some(time_exchange) = millis_to_datetime(fill.time) else {
                warn!(time = fill.time, "Invalid fill timestamp, skipping");
                continue;
            };

            result.push(Trade {
                id: TradeId(format_smolstr!("{}", fill.tid)),
                order_id: OrderId(format_smolstr!("{}", fill.oid)),
                instrument,
                strategy: StrategyId::unknown(),
                time_exchange,
                side,
                price,
                quantity,
                // `TradeInfo` carries no cumulative filled quantity; Hyperliquid reports order
                // state as its own `OrderUpdate` message.
                order_filled_quantity: None,
                fees: AssetFees {
                    asset: AssetNameExchange::from("USDC"),
                    fees: fee,
                    fees_quote: Some(fee),
                },
            });
        }

        Ok(result)
    }
}

/// Convert SDK TradeInfo (fill) to AccountEvent::Trade.
fn fill_to_account_event(fill: &hyperliquid_rust_sdk::TradeInfo) -> Option<UnindexedAccountEvent> {
    let side = parse_side(&fill.side)?;
    let price = parse_decimal(&fill.px, "fill.px")?;
    let quantity = parse_decimal(&fill.sz, "fill.sz")?;
    let fee = parse_decimal(&fill.fee, "fill.fee").unwrap_or(Decimal::ZERO);
    let time_exchange = millis_to_datetime(fill.time)?;
    let instrument = perp_coin_to_instrument(&fill.coin);
    let order_id = OrderId(format_smolstr!("{}", fill.oid));

    let trade = Trade {
        id: TradeId(format_smolstr!("{}", fill.tid)),
        order_id,
        instrument,
        strategy: StrategyId::unknown(),
        time_exchange,
        side,
        price,
        quantity,
        // `TradeInfo` carries no cumulative filled quantity, so the order's state must be learned
        // from an `OrderUpdate` -- which Hyperliquid sends as its own message.
        order_filled_quantity: None,
        fees: AssetFees {
            asset: AssetNameExchange::from("USDC"),
            fees: fee,
            fees_quote: Some(fee),
        },
    };

    Some(AccountEvent::new(
        ExchangeId::HyperliquidPerp,
        AccountEventKind::Trade(trade),
    ))
}

/// Convert SDK OrderUpdate to AccountEvent::OrderSnapshot.
fn order_update_to_account_event(
    update: &hyperliquid_rust_sdk::OrderUpdate,
) -> Option<UnindexedAccountEvent> {
    let order = &update.order;
    let side = parse_side(&order.side)?;
    let price = parse_decimal(&order.limit_px, "order.limit_px")?;
    let orig_sz = parse_decimal(&order.orig_sz, "order.orig_sz")?;
    let time_exchange = millis_to_datetime(update.status_timestamp)?;
    let instrument = perp_coin_to_instrument(&order.coin);

    // Use cloid (client order ID) if available, fall back to OID
    let order_id_smol = format_smolstr!("{}", order.oid);
    let cid = order
        .cloid
        .as_deref()
        .map(|c| ClientOrderId::new(SmolStr::new(c)))
        .unwrap_or_else(|| ClientOrderId::new(order_id_smol.clone()));

    // Determine order state from status
    let state = match update.status.as_str() {
        "open" | "resting" => {
            let current_sz = parse_decimal(&order.sz, "order.sz")?;
            let filled_quantity = (orig_sz - current_sz).max(Decimal::ZERO);
            crate::order::state::OrderState::active(Open {
                id: OrderId(order_id_smol),
                time_exchange,
                filled_quantity,
            })
        }
        "filled" => crate::order::state::OrderState::fully_filled(Filled::new(
            OrderId(order_id_smol),
            time_exchange,
            orig_sz, // Fully filled means filled_quantity == orig_sz
            None,    // OrderUpdate doesn't include avg_price
        )),
        "canceled" | "cancelled" => {
            let current_sz = parse_decimal(&order.sz, "order.sz")?;
            let filled_quantity = (orig_sz - current_sz).max(Decimal::ZERO);
            crate::order::state::OrderState::inactive(Cancelled::new(
                OrderId(order_id_smol),
                time_exchange,
                filled_quantity,
            ))
        }
        status => {
            warn!(%status, "Unknown order status");
            return None;
        }
    };

    // SDK's OrderUpdate doesn't include original order type or TIF, so we default
    // to Limit/GTC. This is a known limitation — IOC/FOK orders will be misrepresented.
    let order_snapshot = Order {
        key: OrderKey {
            exchange: ExchangeId::HyperliquidPerp,
            instrument,
            strategy: StrategyId::unknown(),
            cid,
        },
        side,
        price: Some(price),
        quantity: orig_sz,
        kind: OrderKind::Limit,
        time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
        state,
    };

    Some(AccountEvent::new(
        ExchangeId::HyperliquidPerp,
        AccountEventKind::OrderSnapshot(Snapshot(order_snapshot)),
    ))
}

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

    #[test]
    fn test_fill_to_account_event() {
        let fill_json = r#"{
            "coin": "BTC",
            "side": "B",
            "px": "65000.5",
            "sz": "0.1",
            "time": 1714100000000,
            "hash": "0xabc123",
            "startPosition": "0",
            "dir": "Open Long",
            "closedPnl": "0",
            "oid": 12345,
            "cloid": null,
            "crossed": false,
            "fee": "0.65",
            "feeToken": "USDC",
            "tid": 99999
        }"#;

        let fill: hyperliquid_rust_sdk::TradeInfo = serde_json::from_str(fill_json).unwrap();
        let event = fill_to_account_event(&fill).unwrap();

        assert_eq!(event.exchange, ExchangeId::HyperliquidPerp);
        match event.kind {
            AccountEventKind::Trade(trade) => {
                assert_eq!(trade.instrument.as_ref(), "BTC-USD-PERP");
                assert_eq!(
                    trade.id.0, "99999",
                    "the id is the fill's tid, not its hash"
                );
                assert_eq!(trade.side, Side::Buy);
                assert_eq!(trade.price, dec!(65000.5));
                assert_eq!(trade.quantity, dec!(0.1));
                assert_eq!(trade.fees.fees, dec!(0.65));
            }
            _ => panic!("Expected Trade event"),
        }
    }

    #[test]
    fn test_fill_to_account_event_sell() {
        let fill_json = r#"{
            "coin": "ETH",
            "side": "A",
            "px": "3200",
            "sz": "1.5",
            "time": 1714100000000,
            "hash": "0xdef456",
            "startPosition": "1.5",
            "dir": "Close Long",
            "closedPnl": "150.0",
            "oid": 12346,
            "cloid": null,
            "crossed": true,
            "fee": "4.8",
            "feeToken": "USDC",
            "tid": 100000
        }"#;

        let fill: hyperliquid_rust_sdk::TradeInfo = serde_json::from_str(fill_json).unwrap();
        let event = fill_to_account_event(&fill).unwrap();

        match event.kind {
            AccountEventKind::Trade(trade) => {
                assert_eq!(trade.instrument.as_ref(), "ETH-USD-PERP");
                assert_eq!(trade.side, Side::Sell);
                assert_eq!(trade.price, dec!(3200));
                assert_eq!(trade.quantity, dec!(1.5));
            }
            _ => panic!("Expected Trade event"),
        }
    }

    /// Build a perp `TradeInfo` sharing one transaction hash with its siblings.
    fn sweep_fill(tid: u64, px: &str) -> hyperliquid_rust_sdk::TradeInfo {
        let json = format!(
            r#"{{
                "coin": "BTC", "side": "B", "px": "{px}", "sz": "0.1",
                "time": 1714100000000, "hash": "0xonesweep", "startPosition": "0",
                "dir": "Open Long", "closedPnl": "0", "oid": 4242, "cloid": null,
                "crossed": true, "fee": "0.65", "feeToken": "USDC", "tid": {tid}
            }}"#
        );
        serde_json::from_str(&json).unwrap()
    }

    #[test]
    fn a_trade_id_identifies_the_fill_not_the_transaction() {
        // One aggressive order sweeping two resting levels: same `hash`, same `oid`, two fills.
        // Deriving the id from `hash` gave both the same `TradeId`, so anything reconciling on it
        // treated them as one fill and dropped the second -- understating filled quantity and fees.
        let first = fill_to_account_event(&sweep_fill(111, "65000.5")).unwrap();
        let second = fill_to_account_event(&sweep_fill(222, "65001.0")).unwrap();

        let (AccountEventKind::Trade(first), AccountEventKind::Trade(second)) =
            (first.kind, second.kind)
        else {
            panic!("Expected Trade events");
        };

        assert_eq!(first.order_id, second.order_id, "one order produced both");
        assert_ne!(first.id, second.id, "but they are distinct fills");
        assert_eq!(first.id.0, "111");
        assert_eq!(second.id.0, "222");
    }

    #[test]
    fn a_replayed_fill_is_delivered_once() {
        // The `userFills` subscription opens with a snapshot of recent fills, and the SDK
        // resubscribes on reconnect -- so the same fill arrives again on every reconnect.
        let cache = new_dedup_cache();
        let event = fill_to_account_event(&sweep_fill(111, "65000.5")).unwrap();
        let replay = fill_to_account_event(&sweep_fill(111, "65000.5")).unwrap();

        assert!(!is_duplicate(&cache, dedup_key_from_event(&event).unwrap()));
        assert!(is_duplicate(&cache, dedup_key_from_event(&replay).unwrap()));
    }

    #[test]
    fn both_fills_of_one_sweep_survive_dedup() {
        // The guard against the fix and the dedup fighting each other: two fills of one sweep are
        // not duplicates of one another, and dedup must not collapse what the id now separates.
        let cache = new_dedup_cache();
        let first = fill_to_account_event(&sweep_fill(111, "65000.5")).unwrap();
        let second = fill_to_account_event(&sweep_fill(222, "65001.0")).unwrap();

        assert!(!is_duplicate(&cache, dedup_key_from_event(&first).unwrap()));
        assert!(!is_duplicate(
            &cache,
            dedup_key_from_event(&second).unwrap()
        ));
    }

    #[test]
    fn one_tid_on_two_instruments_is_two_fills() {
        // Hyperliquid documents `tid` as qualified by coin rather than globally unique, which is
        // why the dedup key carries the instrument.
        let cache = new_dedup_cache();
        let btc = fill_to_account_event(&sweep_fill(111, "65000.5")).unwrap();

        let eth_json = r#"{
            "coin": "ETH", "side": "B", "px": "3200", "sz": "1.5",
            "time": 1714100000000, "hash": "0xother", "startPosition": "0",
            "dir": "Open Long", "closedPnl": "0", "oid": 9, "cloid": null,
            "crossed": true, "fee": "4.8", "feeToken": "USDC", "tid": 111
        }"#;
        let eth: hyperliquid_rust_sdk::TradeInfo = serde_json::from_str(eth_json).unwrap();
        let eth = fill_to_account_event(&eth).unwrap();

        assert!(!is_duplicate(&cache, dedup_key_from_event(&btc).unwrap()));
        assert!(!is_duplicate(&cache, dedup_key_from_event(&eth).unwrap()));
    }

    #[test]
    fn test_order_update_to_account_event_open() {
        let update_json = r#"{
            "order": {
                "coin": "BTC",
                "side": "B",
                "limitPx": "64000",
                "sz": "0.5",
                "oid": 12345,
                "timestamp": 1714100000000,
                "origSz": "0.5",
                "cloid": null
            },
            "status": "open",
            "statusTimestamp": 1714100000000
        }"#;

        let update: hyperliquid_rust_sdk::OrderUpdate = serde_json::from_str(update_json).unwrap();
        let event = order_update_to_account_event(&update).unwrap();

        assert_eq!(event.exchange, ExchangeId::HyperliquidPerp);
        match event.kind {
            AccountEventKind::OrderSnapshot(Snapshot(order)) => {
                assert_eq!(order.key.instrument.as_ref(), "BTC-USD-PERP");
                assert_eq!(order.side, Side::Buy);
                assert_eq!(order.price, Some(dec!(64000)));
                assert_eq!(order.quantity, dec!(0.5));
                assert!(matches!(
                    order.state,
                    crate::order::state::OrderState::Active(_)
                ));
            }
            _ => panic!("Expected OrderSnapshot event"),
        }
    }

    #[test]
    fn test_order_update_to_account_event_filled() {
        let update_json = r#"{
            "order": {
                "coin": "ETH",
                "side": "A",
                "limitPx": "3250",
                "sz": "0",
                "oid": 12346,
                "timestamp": 1714100000000,
                "origSz": "2.0",
                "cloid": null
            },
            "status": "filled",
            "statusTimestamp": 1714100001000
        }"#;

        let update: hyperliquid_rust_sdk::OrderUpdate = serde_json::from_str(update_json).unwrap();
        let event = order_update_to_account_event(&update).unwrap();

        match event.kind {
            AccountEventKind::OrderSnapshot(Snapshot(order)) => {
                assert_eq!(order.side, Side::Sell);
                assert!(matches!(
                    order.state,
                    crate::order::state::OrderState::Inactive(
                        crate::order::state::InactiveOrderState::FullyFilled(_)
                    )
                ));
            }
            _ => panic!("Expected OrderSnapshot event"),
        }
    }

    #[test]
    fn test_order_update_to_account_event_cancelled() {
        let update_json = r#"{
            "order": {
                "coin": "SOL",
                "side": "B",
                "limitPx": "150",
                "sz": "10",
                "oid": 12347,
                "timestamp": 1714100000000,
                "origSz": "10",
                "cloid": null
            },
            "status": "canceled",
            "statusTimestamp": 1714100002000
        }"#;

        let update: hyperliquid_rust_sdk::OrderUpdate = serde_json::from_str(update_json).unwrap();
        let event = order_update_to_account_event(&update).unwrap();

        match event.kind {
            AccountEventKind::OrderSnapshot(Snapshot(order)) => {
                assert_eq!(order.key.instrument.as_ref(), "SOL-USD-PERP");
                assert!(matches!(
                    order.state,
                    crate::order::state::OrderState::Inactive(
                        crate::order::state::InactiveOrderState::Cancelled(_)
                    )
                ));
            }
            _ => panic!("Expected OrderSnapshot event"),
        }
    }

    #[test]
    fn test_order_update_unknown_status_returns_none() {
        let update_json = r#"{
            "order": {
                "coin": "BTC",
                "side": "B",
                "limitPx": "64000",
                "sz": "0.5",
                "oid": 12345,
                "timestamp": 1714100000000,
                "origSz": "0.5",
                "cloid": null
            },
            "status": "unknown_status",
            "statusTimestamp": 1714100000000
        }"#;

        let update: hyperliquid_rust_sdk::OrderUpdate = serde_json::from_str(update_json).unwrap();
        let event = order_update_to_account_event(&update);
        assert!(event.is_none());
    }
}