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.
//!
//! # Client order ids
//!
//! **Every order must be placed under a client id in
//! [`ClientOrderId::uuid()`](crate::order::id::ClientOrderId::uuid) form**: a UUID, lowercase and
//! hyphenated. Any other id is refused at placement with [`OrderError::Rejected`], before anything
//! is sent.
//!
//! Hyperliquid tracks an order by its venue `oid` and by an optional 16-byte `cloid`, and reports
//! the `cloid` back on every order record, REST and WebSocket alike. This client sends the client
//! id as the `cloid` and turns the one reported back into the same id, so every account snapshot,
//! open-order listing and order update names an order by the id it was placed with. Only that
//! form of UUID comes back unchanged, which is why nothing else is accepted.
//!
//! An order placed some other way, such as the web app, has no `cloid` and is reported under its
//! `oid`. It cannot be one of this client's orders, which is what lets an account snapshot declare
//! its order list complete (see [`InstrumentAccountSnapshot::orders_complete`]).
//!
//! # 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)
//!
//! **SDK limitations** (hyperliquid_rust_sdk 0.6.x): neither the open orders this client reads
//! nor `OrderUpdate` carry an order's trigger fields or time in force, so orders from both are
//! reported as `OrderKind::Limit`, `GoodUntilCancelled`. Track `OrderKind` from the placement
//! request.
//!
//! # 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::{OrderId, StrategyId, VenueOrderId},
        request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
        state::{Filled, Open, OrderState, UnindexedOrderState},
    },
    position::{Position, PositionReport},
    trade::{AssetFees, Trade, TradeId},
};
use chrono::{DateTime, Utc};
use common::{
    CLOID_REQUIRED, CancelOnDropStream, OpenOrderListing, cid_to_cloid, instrument_to_perp_coin,
    map_tif, millis_to_datetime, open_order_to_order, open_orders, 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 smol_str::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),
        }
    }

    /// The USDC collateral balance, and each perpetual's open orders and position.
    ///
    /// # Positions
    ///
    /// Every requested perpetual is listed. Its position is [`PositionReport::Open`] when
    /// Hyperliquid reports a non-zero size for it, and [`PositionReport::Flat`] otherwise, since
    /// the user state holds every open perpetual position. A size that does not parse is
    /// [`PositionReport::Unreported`], with a warning.
    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)
            },
            open_orders(&self.info_client, address)
        )?;

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

        let mut listing = OpenOrderListing::new(
            &open_orders,
            ExchangeId::HyperliquidPerp,
            instruments,
            |coin| Some(perp_coin_to_instrument(coin)),
        );

        // 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 position = perp_position_report(pos, now);

            let (orders, orders_complete) = match listing.remove(&instrument) {
                Some(listed) => (listed.orders, listed.orders_complete),
                // The listing has an entry for every requested instrument, so this is an
                // unfiltered read with no open order on the instrument.
                None => (Vec::new(), true),
            };

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

        // Every other instrument listed: open orders but no position, or requested with neither.
        // `asset_positions` holds every open perp position, so these are flat.
        instrument_snapshots.extend(listing.into_snapshots().map(|mut snapshot| {
            snapshot.position = PositionReport::Flat;
            snapshot
        }));

        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);

        enum CancelMethod {
            ByOid(u64),
            ByCloid(Uuid),
        }

        // How the venue addresses the order decides which endpoint cancels it. This reads the
        // identifier's kind rather than inferring it from the string's shape: an oid and a cloid
        // are different identifiers that happen to both be text, and telling them apart by parsing
        // means a cloid which parses as a number gets cancelled as somebody else's oid.
        let cancel_method = match &request.state.id {
            Some(VenueOrderId::Assigned(id)) => {
                id.0.parse::<u64>()
                    .map(CancelMethod::ByOid)
                    .map_err(|_| "venue order id is not a numeric Hyperliquid oid")
            }
            // The venue accepted the order without assigning an oid, so the only handle on it is
            // the cloid it was placed with -- which placement derives from the client id the same
            // way, and refuses the order if it cannot.
            Some(VenueOrderId::ClientAssigned) => cid_to_cloid(&request.key.cid)
                .map(CancelMethod::ByCloid)
                .ok_or("order is addressable only by its client id, which is not a canonical UUID"),
            None => Err("order is still in flight; the venue has acknowledged nothing to cancel"),
        };

        let cancel_method = match cancel_method {
            Ok(method) => method,
            Err(reason) => {
                warn!(%reason, cid = %request.key.cid, "Cannot determine how to cancel order");
                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(reason.to_string()),
                    )),
                });
            }
        };

        // Terminal states keep a plain `OrderId`: nothing compares them for identity, so an order
        // the venue never named records the client id it was addressed by instead.
        let cancelled_id = match &request.state.id {
            Some(venue_id) => venue_id.or_client_id(&request.key.cid),
            None => OrderId(request.key.cid.0.clone()),
        };

        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 (no oid assigned)");
                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(
                        cancelled_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 { .. } => {}
        }

        // The venue reports every order under its cloid, so an order placed without one could
        // not be matched to the id the caller tracks it by. See the module docs.
        let Some(cloid) = cid_to_cloid(&request.key.cid) else {
            return Some(make_rejected(CLOID_REQUIRED.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: Some(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: VenueOrderId::Assigned(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,
                    ) => {
                        // The venue answered without an oid, so the order's only handle is the
                        // cloid it was placed with. A standalone trigger order is answered
                        // `resting` with an oid; these statuses belong to grouped orders.
                        debug!(cid = %request.key.cid, "Order waiting under its cloid");
                        OrderState::active(Open {
                            id: VenueOrderId::ClientAssigned,
                            time_exchange: Utc::now(),
                            filled_quantity: Decimal::ZERO,
                        })
                    }
                    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 = open_orders(&self.info_client, address).await?;

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

        Ok(open_orders
            .iter()
            .filter_map(|order| {
                let instrument = perp_coin_to_instrument(&order.coin);
                if instrument_filter
                    .as_ref()
                    .is_some_and(|f| !f.contains(&instrument))
                {
                    return None;
                }
                open_order_to_order(order, ExchangeId::HyperliquidPerp, instrument)
            })
            .collect())
    }

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

/// Report one perpetual position from Hyperliquid's user state.
///
/// [`PositionReport::Open`] for a non-zero size, [`PositionReport::Flat`] for zero, and
/// [`PositionReport::Unreported`] when the size does not parse (with a warning): an unreadable
/// size establishes nothing, so it must not read as flat.
fn perp_position_report(
    position: &hyperliquid_rust_sdk::PositionData,
    now: DateTime<Utc>,
) -> PositionReport {
    let Some(quantity) = parse_decimal(&position.szi, "szi") else {
        return PositionReport::Unreported;
    };
    PositionReport::from_position(Position::new(
        quantity,
        position
            .entry_px
            .as_ref()
            .and_then(|p| parse_decimal(p, "entry_px")),
        parse_decimal(&position.unrealized_pnl, "unrealized_pnl"),
        parse_decimal(&position.margin_used, "margin_used"),
        position
            .liquidation_px
            .as_ref()
            .and_then(|p| parse_decimal(p, "liquidation_px")),
        Some(Decimal::from(position.leverage.value)),
        now,
    ))
}

/// 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> {
    common::order_update_to_account_event(
        update,
        ExchangeId::HyperliquidPerp,
        perp_coin_to_instrument(&update.order.coin),
    )
}

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

    fn perp_position(szi: &str) -> hyperliquid_rust_sdk::PositionData {
        serde_json::from_value(serde_json::json!({
            "coin": "BTC",
            "entryPx": "50000.0",
            "leverage": {"type": "cross", "value": 5},
            "liquidationPx": null,
            "marginUsed": "100.0",
            "positionValue": "500.0",
            "returnOnEquity": "0.0",
            "szi": szi,
            "unrealizedPnl": "1.5",
            "maxLeverage": 50,
            "cumFunding": {"allTime": "0", "sinceOpen": "0", "sinceChange": "0"}
        }))
        .unwrap()
    }

    #[test]
    fn perp_position_report_is_open_flat_or_unreported() {
        let now = Utc::now();
        let open = perp_position_report(&perp_position("-0.01"), now);
        let position = open.open().unwrap();
        assert_eq!(
            (position.quantity, position.entry_price, position.leverage),
            (dec!(-0.01), Some(dec!(50000.0)), Some(dec!(5)))
        );
        assert_eq!(
            perp_position_report(&perp_position("0.0"), now),
            PositionReport::Flat
        );
        assert_eq!(
            perp_position_report(&perp_position("not a number"), now),
            PositionReport::Unreported
        );
    }

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